@cryptotaxi247 / netdata-1 / commits / 7f66adcf3

netflow-plugin: isolate dependency upgrades (#22273)

* netflow-plugin: isolate dependency upgrades * netflow-plugin: keep serde_yaml instead of switching to serde_yaml_ng Revert the yaml crate swap from the dependency-upgrade commit. There is no published RustSec advisory (or CVE) against serde_yaml today, only an open proposal, so the swap silenced no warning and added a second YAML parser to the workspace alongside the one still used by otel-plugin and otel-signal-viewer-plugin. * netflow-plugin: lift dep upgrades to workspace pins The previous commit pinned each upgraded dependency directly inside netflow-plugin/Cargo.toml. Move those bumps up to [workspace.dependencies] so the workspace remains the single source of truth, and switch the netflow-plugin entries back to { workspace = true }. * deps: bump bytesize to 2.x workspace-wide and drop bytesize-serde bytesize 2.3.1 ships its own serde impl behind the "serde" feature. It serializes ByteSize as a string ("100MB") in human-readable formats and accepts both strings and integers on deserialization, so existing YAML config files keep parsing unchanged -- and integer byte counts now also work as a bonus. * deps: drop dead bincode workspace pin The bincode workspace pin had only one consumer (netflow-plugin), which moved to rmp-serde a few commits back. No other workspace crate declares bincode in its Cargo.toml or imports it in source. * netflow-plugin: size-check facet state load, share jaq compile, type-check GeoIP lookups facet_runtime: pre-check facet state file size before reading. Mirrors the existing pattern in ingest/persistence.rs so an oversized or corrupt state file is rejected via fs::metadata() rather than a full fs::read() into memory. Adds MAX_FACET_STATE_FILE_LEN and a unit test using a sparse file. network_sources/transform + plugin_config/validation/enrichment: factor the duplicated jaq compile pipeline (defs + funs + Loader + Compiler) into a shared compile_jaq_filter exported pub(crate) from network_sources, used by both runtime compilation and config validation. Replace the {:?} debug formatting of jaq parse/compile errors with a renderer that produces a joined Display string covering Io/Lex/Parse and Undefined symbol cases, truncating long input fragments for readability. enrichment/data/geoip/resolver: detect IPv6 lookups against IPv4-only databases via maxminddb Reader.metadata.ip_version (publicly exposed in 0.27.x) before calling lookup(), instead of pattern-matching on an upstream English error message. Removes the brittle string match and the test that exercised it. --------- Co-authored-by: vkalintiris <vasilis@netdata.cloud>

Costa Tsaousis committed Apr 26, 2026 at 01:42 UTC 7f66adcf3ccf9b46a8fc8bf2773e398c0cd20bba
36 files changed +832 -264
src/crates/Cargo.lock
+420 -70
@@ -209,6 +209,28 @@ version = "1.5.0"
209 source = "registry+https://github.com/rust-lang/crates.io-index"
210 checksum = "c08606f8c3cbf4ce6ec8e28fb0014a2c086708fe954eaa885384a6165172e7e8"
211
212 +[[package]]
213 +name = "aws-lc-rs"
214 +version = "1.16.3"
215 +source = "registry+https://github.com/rust-lang/crates.io-index"
216 +checksum = "0ec6fb3fe69024a75fa7e1bfb48aa6cf59706a101658ea01bfd33b2b248a038f"
217 +dependencies = [
218 + "aws-lc-sys",
219 + "zeroize",
220 +]
221 +
222 +[[package]]
223 +name = "aws-lc-sys"
224 +version = "0.40.0"
225 +source = "registry+https://github.com/rust-lang/crates.io-index"
226 +checksum = "f50037ee5e1e41e7b8f9d161680a725bd1626cb6f8c7e901f91f942850852fe7"
227 +dependencies = [
228 + "cc",
229 + "cmake",
230 + "dunce",
231 + "fs_extra",
232 +]
233 +
234 [[package]]
235 name = "axum"
236 version = "0.7.9"
@@ -390,6 +412,17 @@ dependencies = [
412 "tracing-subscriber",
413 ]
414
415 +[[package]]
416 +name = "bstr"
417 +version = "1.12.1"
418 +source = "registry+https://github.com/rust-lang/crates.io-index"
419 +checksum = "63044e1ae8e69f3b5a92c736ca6269b8d12fa7efe39bf34ddb06d102cf0e2cab"
420 +dependencies = [
421 + "memchr",
422 + "regex-automata",
423 + "serde",
424 +]
425 +
426 [[package]]
427 name = "bumpalo"
428 version = "3.20.2"
@@ -425,18 +458,11 @@ checksum = "1e748733b7cbc798e1434b6ac524f0c1ff2ab456fe201501e6497c8417a4fc33"
458
459 [[package]]
460 name = "bytesize"
428 -version = "1.3.3"
461 +version = "2.3.1"
462 source = "registry+https://github.com/rust-lang/crates.io-index"
430 -checksum = "2e93abca9e28e0a1b9877922aacb20576e05d4679ffa78c3d6dc22a26a216659"
431 -
432 -[[package]]
433 -name = "bytesize-serde"
434 -version = "0.2.1"
435 -source = "registry+https://github.com/rust-lang/crates.io-index"
436 -checksum = "86d1eb2fd2668859e9785b99700c66b7ea9fdeda35d99f95827a4e5493440178"
463 +checksum = "6bd91ee7b2422bcb158d90ef4d14f75ef67f340943fc4149891dcce8f8b972a3"
464 dependencies = [
438 - "bytesize",
439 - "serde",
465 + "serde_core",
466 ]
467
468 [[package]]
@@ -451,6 +477,12 @@ dependencies = [
477 "shlex",
478 ]
479
480 +[[package]]
481 +name = "cesu8"
482 +version = "1.1.0"
483 +source = "registry+https://github.com/rust-lang/crates.io-index"
484 +checksum = "6d43a04d8753f35258c91f8ec639f792891f748a1edbd759cf1dcea3382ad83c"
485 +
486 [[package]]
487 name = "cfg-if"
488 version = "1.0.4"
@@ -517,6 +549,15 @@ version = "1.1.0"
549 source = "registry+https://github.com/rust-lang/crates.io-index"
550 checksum = "c8d4a3bb8b1e0c1050499d1815f5ab16d04f0959b233085fb31653fbfc9d98f9"
551
552 +[[package]]
553 +name = "cmake"
554 +version = "0.1.58"
555 +source = "registry+https://github.com/rust-lang/crates.io-index"
556 +checksum = "c0f78a02292a74a88ac736019ab962ece0bc380e3f977bf72e376c5d78ff0678"
557 +dependencies = [
558 + "cc",
559 +]
560 +
561 [[package]]
562 name = "cmsketch"
563 version = "0.2.4"
@@ -529,6 +570,16 @@ version = "1.0.5"
570 source = "registry+https://github.com/rust-lang/crates.io-index"
571 checksum = "1d07550c9036bf2ae0c684c4297d503f838287c83c53686d05370d0e139ae570"
572
573 +[[package]]
574 +name = "combine"
575 +version = "4.6.7"
576 +source = "registry+https://github.com/rust-lang/crates.io-index"
577 +checksum = "ba5a308b75df32fe02788e748662718f03fde005016435c444eea572398219fd"
578 +dependencies = [
579 + "bytes",
580 + "memchr",
581 +]
582 +
583 [[package]]
584 name = "concurrent-queue"
585 version = "2.5.0"
@@ -776,6 +827,12 @@ version = "1.2.1"
827 source = "registry+https://github.com/rust-lang/crates.io-index"
828 checksum = "75b325c5dbd37f80359721ad39aca5a29fb04c89279657cffdda8736d0c0b9d2"
829
830 +[[package]]
831 +name = "dunce"
832 +version = "1.0.5"
833 +source = "registry+https://github.com/rust-lang/crates.io-index"
834 +checksum = "92773504d58c093f6de2459af4af33faa518c13451eb8f2b5698ed3d36e7c813"
835 +
836 [[package]]
837 name = "dyn-clone"
838 version = "1.0.20"
@@ -806,9 +863,9 @@ dependencies = [
863
864 [[package]]
865 name = "etherparse"
809 -version = "0.19.0"
866 +version = "0.20.1"
867 source = "registry+https://github.com/rust-lang/crates.io-index"
811 -checksum = "b119b9796ff800751a220394b8b3613f26dd30c48f254f6837e64c464872d1c7"
868 +checksum = "3ac016aaf11dfe643edcd088a166234bfcb72e7f06691abfcf21e9af524037f4"
869 dependencies = [
870 "arrayvec",
871 ]
@@ -1046,6 +1103,12 @@ dependencies = [
1103 "windows-sys 0.59.0",
1104 ]
1105
1106 +[[package]]
1107 +name = "fs_extra"
1108 +version = "1.3.0"
1109 +source = "registry+https://github.com/rust-lang/crates.io-index"
1110 +checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c"
1111 +
1112 [[package]]
1113 name = "fsevent-sys"
1114 version = "4.1.0"
@@ -1261,6 +1324,17 @@ dependencies = [
1324 "foldhash 0.2.0",
1325 ]
1326
1327 +[[package]]
1328 +name = "hashbrown"
1329 +version = "0.17.0"
1330 +source = "registry+https://github.com/rust-lang/crates.io-index"
1331 +checksum = "4f467dd6dccf739c208452f8014c75c18bb8301b050ad1cfb27153803edb0f51"
1332 +dependencies = [
1333 + "allocator-api2",
1334 + "equivalent",
1335 + "foldhash 0.2.0",
1336 +]
1337 +
1338 [[package]]
1339 name = "hashers"
1340 version = "1.0.1"
@@ -1297,9 +1371,9 @@ checksum = "fc0fef456e4baa96da950455cd02c081ca953b141298e41db3fc7e36b1da849c"
1371
1372 [[package]]
1373 name = "hifijson"
1300 -version = "0.2.3"
1374 +version = "0.5.0"
1375 source = "registry+https://github.com/rust-lang/crates.io-index"
1302 -checksum = "0a7763b98ba8a24f59e698bf9ab197e7676c640d6455d1580b4ce7dc560f0f0d"
1376 +checksum = "242402749acf71e6f32f5857598b7002c4058a4e3c3b22b4c7d51cab9aea754e"
1377
1378 [[package]]
1379 name = "http"
@@ -1398,7 +1472,6 @@ dependencies = [
1472 "tokio",
1473 "tokio-rustls",
1474 "tower-service",
1401 - "webpki-roots",
1475 ]
1476
1477 [[package]]
@@ -1687,52 +1760,135 @@ checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682"
1760
1761 [[package]]
1762 name = "jaq-core"
1690 -version = "1.5.1"
1763 +version = "3.0.0"
1764 +source = "registry+https://github.com/rust-lang/crates.io-index"
1765 +checksum = "dca0f164c8e9c55fc5aefe60b371df735c719b09930dac185878f2f8c7ab6b68"
1766 +dependencies = [
1767 + "dyn-clone",
1768 + "once_cell",
1769 + "typed-arena",
1770 +]
1771 +
1772 +[[package]]
1773 +name = "jaq-json"
1774 +version = "2.0.0"
1775 +source = "registry+https://github.com/rust-lang/crates.io-index"
1776 +checksum = "f5c5baabe63d1d72cde60ec7548a098036773e3541dbc65c6a44fb38e9cfb272"
1777 +dependencies = [
1778 + "bstr",
1779 + "bytes",
1780 + "foldhash 0.1.5",
1781 + "hifijson",
1782 + "indexmap 2.13.1",
1783 + "jaq-core",
1784 + "jaq-std",
1785 + "num-bigint",
1786 + "num-traits",
1787 + "ryu",
1788 + "self_cell",
1789 + "serde_core",
1790 +]
1791 +
1792 +[[package]]
1793 +name = "jaq-std"
1794 +version = "3.0.0"
1795 source = "registry+https://github.com/rust-lang/crates.io-index"
1692 -checksum = "d6fda09ee08c84c81293fdf811d9ebaa87b327557b5391f290c926d728c2ddd4"
1796 +checksum = "2a11bb307027b20b3dc7b212ad687e7e410cbc43933eec5d498672ab2bc60666"
1797 dependencies = [
1798 "aho-corasick",
1799 "base64 0.22.1",
1696 - "chrono",
1697 - "hifijson",
1698 - "jaq-interpret",
1800 + "bstr",
1801 + "jaq-core",
1802 + "jiff",
1803 "libm",
1804 "log",
1701 - "regex",
1805 + "regex-bites",
1806 "urlencoding",
1807 ]
1808
1809 [[package]]
1706 -name = "jaq-interpret"
1707 -version = "1.5.0"
1810 +name = "jiff"
1811 +version = "0.2.24"
1812 source = "registry+https://github.com/rust-lang/crates.io-index"
1709 -checksum = "2fe95ec3c24af3fd9f3dd1091593f5e49b003a66c496a8aa39d764d0a06ae17b"
1813 +checksum = "f00b5dbd620d61dfdcb6007c9c1f6054ebd75319f163d886a9055cec1155073d"
1814 dependencies = [
1711 - "ahash",
1712 - "dyn-clone",
1713 - "hifijson",
1714 - "indexmap 2.13.1",
1715 - "jaq-syn",
1716 - "once_cell",
1717 - "serde_json",
1815 + "jiff-static",
1816 + "jiff-tzdb-platform",
1817 + "log",
1818 + "portable-atomic",
1819 + "portable-atomic-util",
1820 + "serde_core",
1821 + "windows-sys 0.61.2",
1822 ]
1823
1824 [[package]]
1721 -name = "jaq-std"
1722 -version = "1.6.0"
1825 +name = "jiff-static"
1826 +version = "0.2.24"
1827 source = "registry+https://github.com/rust-lang/crates.io-index"
1724 -checksum = "bfbaa55578fd3b70433b594a370741e0c364e4afff92cc0099623fce87311bc1"
1828 +checksum = "e000de030ff8022ea1da3f466fbb0f3a809f5e51ed31f6dd931c35181ad8e6d7"
1829 dependencies = [
1726 - "jaq-syn",
1830 + "proc-macro2",
1831 + "quote",
1832 + "syn 2.0.117",
1833 ]
1834
1835 [[package]]
1730 -name = "jaq-syn"
1731 -version = "1.6.0"
1836 +name = "jiff-tzdb"
1837 +version = "0.1.6"
1838 +source = "registry+https://github.com/rust-lang/crates.io-index"
1839 +checksum = "c900ef84826f1338a557697dc8fc601df9ca9af4ac137c7fb61d4c6f2dfd3076"
1840 +
1841 +[[package]]
1842 +name = "jiff-tzdb-platform"
1843 +version = "0.1.3"
1844 source = "registry+https://github.com/rust-lang/crates.io-index"
1733 -checksum = "1ba44fe4428c71304604261ecbae047ee9cfb60c4f1a6bd222ebbb31726d3948"
1845 +checksum = "875a5a69ac2bab1a891711cf5eccbec1ce0341ea805560dcd90b7a2e925132e8"
1846 dependencies = [
1735 - "serde",
1847 + "jiff-tzdb",
1848 +]
1849 +
1850 +[[package]]
1851 +name = "jni"
1852 +version = "0.21.1"
1853 +source = "registry+https://github.com/rust-lang/crates.io-index"
1854 +checksum = "1a87aa2bb7d2af34197c04845522473242e1aa17c12f4935d5856491a7fb8c97"
1855 +dependencies = [
1856 + "cesu8",
1857 + "cfg-if",
1858 + "combine",
1859 + "jni-sys 0.3.1",
1860 + "log",
1861 + "thiserror 1.0.69",
1862 + "walkdir",
1863 + "windows-sys 0.45.0",
1864 +]
1865 +
1866 +[[package]]
1867 +name = "jni-sys"
1868 +version = "0.3.1"
1869 +source = "registry+https://github.com/rust-lang/crates.io-index"
1870 +checksum = "41a652e1f9b6e0275df1f15b32661cf0d4b78d4d87ddec5e0c3c20f097433258"
1871 +dependencies = [
1872 + "jni-sys 0.4.1",
1873 +]
1874 +
1875 +[[package]]
1876 +name = "jni-sys"
1877 +version = "0.4.1"
1878 +source = "registry+https://github.com/rust-lang/crates.io-index"
1879 +checksum = "c6377a88cb3910bee9b0fa88d4f42e1d2da8e79915598f65fb0c7ee14c878af2"
1880 +dependencies = [
1881 + "jni-sys-macros",
1882 +]
1883 +
1884 +[[package]]
1885 +name = "jni-sys-macros"
1886 +version = "0.4.1"
1887 +source = "registry+https://github.com/rust-lang/crates.io-index"
1888 +checksum = "38c0b942f458fe50cdac086d2f946512305e5631e720728f2a61aabcd47a6264"
1889 +dependencies = [
1890 + "quote",
1891 + "syn 2.0.117",
1892 ]
1893
1894 [[package]]
@@ -2128,15 +2284,15 @@ checksum = "47e1ffaa40ddd1f3ed91f717a33c8c0ee23fff369e3aa8772b9605cc1d22f4c3"
2284
2285 [[package]]
2286 name = "maxminddb"
2131 -version = "0.25.0"
2287 +version = "0.27.3"
2288 source = "registry+https://github.com/rust-lang/crates.io-index"
2133 -checksum = "144de2546bf4846c6c84b7f76be035f7ebbc1e7d40cfb05810ba45c129508321"
2289 +checksum = "76371bd37ce742f8954daabd0fde7f1594ee43ac2200e20c003ba5c3d65e2192"
2290 dependencies = [
2291 "ipnetwork",
2292 "log",
2293 "memchr",
2138 - "memmap2",
2294 "serde",
2295 + "thiserror 2.0.18",
2296 ]
2297
2298 [[package]]
@@ -2296,25 +2452,22 @@ dependencies = [
2452 "allocative",
2453 "anyhow",
2454 "async-trait",
2299 - "bincode",
2455 "bitflags 2.11.0",
2456 "bitvec",
2457 "bytesize",
2303 - "bytesize-serde",
2458 "chrono",
2459 "clap",
2460 "etherparse",
2461 "fst",
2308 - "hashbrown 0.15.5",
2462 + "hashbrown 0.17.0",
2463 "humantime",
2464 "humantime-serde",
2465 "ipnet",
2466 "ipnet-trie",
2467 "itoa",
2468 "jaq-core",
2315 - "jaq-interpret",
2469 + "jaq-json",
2470 "jaq-std",
2317 - "jaq-syn",
2471 "journal-common",
2472 "journal-core",
2473 "journal-engine",
@@ -2336,7 +2489,8 @@ dependencies = [
2489 "prost 0.14.3",
2490 "protoc-bin-vendored",
2491 "regex",
2339 - "reqwest",
2492 + "reqwest 0.13.2",
2493 + "rmp-serde",
2494 "roaring",
2495 "rt",
2496 "rustc-hash",
@@ -2345,7 +2499,7 @@ dependencies = [
2499 "serde_json",
2500 "serde_yaml",
2501 "sflow-parser",
2348 - "socket2 0.5.10",
2502 + "socket2 0.6.3",
2503 "tempfile",
2504 "thiserror 2.0.18",
2505 "tokio",
@@ -2375,9 +2529,9 @@ dependencies = [
2529
2530 [[package]]
2531 name = "netgauze-bgp-pkt"
2378 -version = "0.9.3"
2532 +version = "0.11.0"
2533 source = "registry+https://github.com/rust-lang/crates.io-index"
2380 -checksum = "976c96f06769f989f41c99f11c06165cfb6b5d9fe0d6fee18d5369804186a590"
2534 +checksum = "5169c61be703b97ba5a1be4606d6696e5c8647e5a2791f837a24011d26124077"
2535 dependencies = [
2536 "byteorder",
2537 "ipnet",
@@ -2392,9 +2546,9 @@ dependencies = [
2546
2547 [[package]]
2548 name = "netgauze-bmp-pkt"
2395 -version = "0.9.3"
2549 +version = "0.11.0"
2550 source = "registry+https://github.com/rust-lang/crates.io-index"
2397 -checksum = "e559ed37796b9d4116a2eb8f6b4bb64ddc143e6e9b30461838248f569badd51b"
2551 +checksum = "0bd1ae6b4400acac3742103f95ffa7688b042f0a36d9adedf255356d20a9adc4"
2552 dependencies = [
2553 "bitflags 2.11.0",
2554 "byteorder",
@@ -2415,9 +2569,9 @@ dependencies = [
2569
2570 [[package]]
2571 name = "netgauze-iana"
2418 -version = "0.9.3"
2572 +version = "0.11.0"
2573 source = "registry+https://github.com/rust-lang/crates.io-index"
2420 -checksum = "d927104bf8eb4c0c4fa62f6089da2ae76cbb6632fa29c712494bf6b00ceeec05"
2574 +checksum = "c5a6d28f40b1aee8cf1bfb9bb6463803a45f87fed99a9762ef56d26ac8852fe5"
2575 dependencies = [
2576 "serde",
2577 "strum_macros",
@@ -2425,18 +2579,18 @@ dependencies = [
2579
2580 [[package]]
2581 name = "netgauze-locate"
2428 -version = "0.9.3"
2582 +version = "0.11.0"
2583 source = "registry+https://github.com/rust-lang/crates.io-index"
2430 -checksum = "ef7b2913ed285eb74b86fde33a6e69525009008f694777b1d892853c17012366"
2584 +checksum = "5ded50ecdaff76ffd41a42708f4b48f2db70f05c3aa0f4f436c3d74beedbe2ee"
2585 dependencies = [
2586 "nom",
2587 ]
2588
2589 [[package]]
2590 name = "netgauze-parse-utils"
2437 -version = "0.9.3"
2591 +version = "0.11.0"
2592 source = "registry+https://github.com/rust-lang/crates.io-index"
2439 -checksum = "d1d87bded02d6ee3d949886f38fe3170df3ea60366c96e69617476532ade293c"
2593 +checksum = "9decd50ea6b044862d048f491a6c8ad62cf2b7d9b905f94e9ddf85d45035d58f"
2594 dependencies = [
2595 "netgauze-locate",
2596 "nom",
@@ -2445,9 +2599,9 @@ dependencies = [
2599
2600 [[package]]
2601 name = "netgauze-serde-macros"
2448 -version = "0.9.3"
2602 +version = "0.11.0"
2603 source = "registry+https://github.com/rust-lang/crates.io-index"
2450 -checksum = "f3c71c31f7611ff298060ea8d1e4f0f3e90390a15ea4d8be52b299253cf2731e"
2604 +checksum = "45f18369c9c01970528485c0fb338040ab86e3433857e362cef20ff17ddd52eb"
2605 dependencies = [
2606 "quote",
2607 "syn 2.0.117",
@@ -2554,6 +2708,25 @@ dependencies = [
2708 "windows-sys 0.61.2",
2709 ]
2710
2711 +[[package]]
2712 +name = "num-bigint"
2713 +version = "0.4.6"
2714 +source = "registry+https://github.com/rust-lang/crates.io-index"
2715 +checksum = "a5e44f723f1133c9deac646763579fdb3ac745e418f2a7af9cd0c431da1f20b9"
2716 +dependencies = [
2717 + "num-integer",
2718 + "num-traits",
2719 +]
2720 +
2721 +[[package]]
2722 +name = "num-integer"
2723 +version = "0.1.46"
2724 +source = "registry+https://github.com/rust-lang/crates.io-index"
2725 +checksum = "7969661fd2958a5cb096e56c8e1ad0444ac2bbcd0061bd28660485a44879858f"
2726 +dependencies = [
2727 + "num-traits",
2728 +]
2729 +
2730 [[package]]
2731 name = "num-traits"
2732 version = "0.2.19"
@@ -2615,7 +2788,7 @@ dependencies = [
2788 "bytes",
2789 "http",
2790 "opentelemetry",
2618 - "reqwest",
2791 + "reqwest 0.12.28",
2792 ]
2793
2794 [[package]]
@@ -2630,7 +2803,7 @@ dependencies = [
2803 "opentelemetry-proto",
2804 "opentelemetry_sdk",
2805 "prost 0.14.3",
2633 - "reqwest",
2806 + "reqwest 0.12.28",
2807 "thiserror 2.0.18",
2808 "tokio",
2809 "tonic 0.14.5",
@@ -2677,7 +2850,6 @@ version = "0.1.3"
2850 dependencies = [
2851 "anyhow",
2852 "bytesize",
2680 - "bytesize-serde",
2853 "clap",
2854 "flatten_otel",
2855 "humantime",
@@ -2709,7 +2881,6 @@ dependencies = [
2881 "anyhow",
2882 "async-trait",
2883 "bytesize",
2712 - "bytesize-serde",
2884 "foyer",
2885 "journal-core",
2886 "journal-function",
@@ -2865,6 +3036,21 @@ version = "0.3.32"
3036 source = "registry+https://github.com/rust-lang/crates.io-index"
3037 checksum = "7edddbd0b52d732b21ad9a5fab5c704c14cd949e5e9a1ec5929a24fded1b904c"
3038
3039 +[[package]]
3040 +name = "portable-atomic"
3041 +version = "1.13.1"
3042 +source = "registry+https://github.com/rust-lang/crates.io-index"
3043 +checksum = "c33a9471896f1c69cecef8d20cbe2f7accd12527ce60845ff44c153bb2a21b49"
3044 +
3045 +[[package]]
3046 +name = "portable-atomic-util"
3047 +version = "0.2.7"
3048 +source = "registry+https://github.com/rust-lang/crates.io-index"
3049 +checksum = "c2a106d1259c23fac8e543272398ae0e3c0b8d33c88ed73d0cc71b0f1d902618"
3050 +dependencies = [
3051 + "portable-atomic",
3052 +]
3053 +
3054 [[package]]
3055 name = "potential_utf"
3056 version = "0.1.5"
@@ -3132,6 +3318,7 @@ version = "0.11.14"
3318 source = "registry+https://github.com/rust-lang/crates.io-index"
3319 checksum = "434b42fec591c96ef50e21e886936e66d3cc3f737104fdb9b737c40ffb94c098"
3320 dependencies = [
3321 + "aws-lc-rs",
3322 "bytes",
3323 "getrandom 0.3.4",
3324 "lru-slab",
@@ -3344,6 +3531,12 @@ dependencies = [
3531 "regex-syntax",
3532 ]
3533
3534 +[[package]]
3535 +name = "regex-bites"
3536 +version = "0.1.6"
3537 +source = "registry+https://github.com/rust-lang/crates.io-index"
3538 +checksum = "b6a15a2fa0bfda9361941c45550896ae87b15cc6c8c939ea350079670332e211"
3539 +
3540 [[package]]
3541 name = "regex-syntax"
3542 version = "0.8.10"
@@ -3365,6 +3558,38 @@ dependencies = [
3558 "http-body",
3559 "http-body-util",
3560 "hyper",
3561 + "hyper-util",
3562 + "js-sys",
3563 + "log",
3564 + "percent-encoding",
3565 + "pin-project-lite",
3566 + "serde",
3567 + "serde_json",
3568 + "serde_urlencoded",
3569 + "sync_wrapper",
3570 + "tokio",
3571 + "tower 0.5.3",
3572 + "tower-http",
3573 + "tower-service",
3574 + "url",
3575 + "wasm-bindgen",
3576 + "wasm-bindgen-futures",
3577 + "web-sys",
3578 +]
3579 +
3580 +[[package]]
3581 +name = "reqwest"
3582 +version = "0.13.2"
3583 +source = "registry+https://github.com/rust-lang/crates.io-index"
3584 +checksum = "ab3f43e3283ab1488b624b44b0e988d0acea0b3214e694730a055cb6b2efa801"
3585 +dependencies = [
3586 + "base64 0.22.1",
3587 + "bytes",
3588 + "futures-core",
3589 + "http",
3590 + "http-body",
3591 + "http-body-util",
3592 + "hyper",
3593 "hyper-rustls",
3594 "hyper-util",
3595 "js-sys",
@@ -3374,9 +3599,9 @@ dependencies = [
3599 "quinn",
3600 "rustls",
3601 "rustls-pki-types",
3602 + "rustls-platform-verifier",
3603 "serde",
3604 "serde_json",
3379 - "serde_urlencoded",
3605 "sync_wrapper",
3606 "tokio",
3607 "tokio-rustls",
@@ -3387,7 +3612,6 @@ dependencies = [
3612 "wasm-bindgen",
3613 "wasm-bindgen-futures",
3614 "web-sys",
3390 - "webpki-roots",
3615 ]
3616
3617 [[package]]
@@ -3404,6 +3628,25 @@ dependencies = [
3628 "windows-sys 0.52.0",
3629 ]
3630
3631 +[[package]]
3632 +name = "rmp"
3633 +version = "0.8.15"
3634 +source = "registry+https://github.com/rust-lang/crates.io-index"
3635 +checksum = "4ba8be72d372b2c9b35542551678538b562e7cf86c3315773cae48dfbfe7790c"
3636 +dependencies = [
3637 + "num-traits",
3638 +]
3639 +
3640 +[[package]]
3641 +name = "rmp-serde"
3642 +version = "1.3.1"
3643 +source = "registry+https://github.com/rust-lang/crates.io-index"
3644 +checksum = "72f81bee8c8ef9b577d1681a70ebbc962c232461e397b22c208c43c04b67a155"
3645 +dependencies = [
3646 + "rmp",
3647 + "serde",
3648 +]
3649 +
3650 [[package]]
3651 name = "roaring"
3652 version = "0.11.2"
@@ -3465,6 +3708,7 @@ version = "0.23.37"
3708 source = "registry+https://github.com/rust-lang/crates.io-index"
3709 checksum = "758025cb5fccfd3bc2fd74708fd4682be41d99e5dff73c377c0646c6012c73a4"
3710 dependencies = [
3711 + "aws-lc-rs",
3712 "log",
3713 "once_cell",
3714 "ring",
@@ -3496,12 +3740,40 @@ dependencies = [
3740 "zeroize",
3741 ]
3742
3743 +[[package]]
3744 +name = "rustls-platform-verifier"
3745 +version = "0.6.2"
3746 +source = "registry+https://github.com/rust-lang/crates.io-index"
3747 +checksum = "1d99feebc72bae7ab76ba994bb5e121b8d83d910ca40b36e0921f53becc41784"
3748 +dependencies = [
3749 + "core-foundation",
3750 + "core-foundation-sys",
3751 + "jni",
3752 + "log",
3753 + "once_cell",
3754 + "rustls",
3755 + "rustls-native-certs",
3756 + "rustls-platform-verifier-android",
3757 + "rustls-webpki",
3758 + "security-framework",
3759 + "security-framework-sys",
3760 + "webpki-root-certs",
3761 + "windows-sys 0.61.2",
3762 +]
3763 +
3764 +[[package]]
3765 +name = "rustls-platform-verifier-android"
3766 +version = "0.1.1"
3767 +source = "registry+https://github.com/rust-lang/crates.io-index"
3768 +checksum = "f87165f0995f63a9fbeea62b64d10b4d9d8e78ec6d7d51fb2125fda7bb36788f"
3769 +
3770 [[package]]
3771 name = "rustls-webpki"
3772 version = "0.103.10"
3773 source = "registry+https://github.com/rust-lang/crates.io-index"
3774 checksum = "df33b2b81ac578cabaf06b89b0631153a3f416b0a886e8a7a1707fb51abbd1ef"
3775 dependencies = [
3776 + "aws-lc-rs",
3777 "ring",
3778 "rustls-pki-types",
3779 "untrusted",
@@ -3612,6 +3884,12 @@ dependencies = [
3884 "libc",
3885 ]
3886
3887 +[[package]]
3888 +name = "self_cell"
3889 +version = "1.2.2"
3890 +source = "registry+https://github.com/rust-lang/crates.io-index"
3891 +checksum = "b12e76d157a900eb52e81bc6e9f3069344290341720e9178cde2407113ac8d89"
3892 +
3893 [[package]]
3894 name = "semver"
3895 version = "1.0.28"
@@ -4377,6 +4655,12 @@ dependencies = [
4655 "rand 0.9.2",
4656 ]
4657
4658 +[[package]]
4659 +name = "typed-arena"
4660 +version = "2.0.2"
4661 +source = "registry+https://github.com/rust-lang/crates.io-index"
4662 +checksum = "6af6ae20167a9ece4bcb41af5b80f8a1f1df981f6391189ce00fd257af04126a"
4663 +
4664 [[package]]
4665 name = "typenum"
4666 version = "1.19.0"
@@ -4635,10 +4919,10 @@ dependencies = [
4919 ]
4920
4921 [[package]]
4638 -name = "webpki-roots"
4639 -version = "1.0.6"
4922 +name = "webpki-root-certs"
4923 +version = "1.0.7"
4924 source = "registry+https://github.com/rust-lang/crates.io-index"
4641 -checksum = "22cfaf3c063993ff62e73cb4311efde4db1efb31ab78a3e5c457939ad5cc0bed"
4925 +checksum = "f31141ce3fc3e300ae89b78c0dd67f9708061d1d2eda54b8209346fd6be9a92c"
4926 dependencies = [
4927 "rustls-pki-types",
4928 ]
@@ -4733,6 +5017,15 @@ dependencies = [
5017 "windows-link",
5018 ]
5019
5020 +[[package]]
5021 +name = "windows-sys"
5022 +version = "0.45.0"
5023 +source = "registry+https://github.com/rust-lang/crates.io-index"
5024 +checksum = "75283be5efb2831d37ea142365f009c02ec203cd29a3ebecbc093d52315b66d0"
5025 +dependencies = [
5026 + "windows-targets 0.42.2",
5027 +]
5028 +
5029 [[package]]
5030 name = "windows-sys"
5031 version = "0.52.0"
@@ -4769,6 +5062,21 @@ dependencies = [
5062 "windows-link",
5063 ]
5064
5065 +[[package]]
5066 +name = "windows-targets"
5067 +version = "0.42.2"
5068 +source = "registry+https://github.com/rust-lang/crates.io-index"
5069 +checksum = "8e5180c00cd44c9b1c88adb3693291f1cd93605ded80c250a75d472756b4d071"
5070 +dependencies = [
5071 + "windows_aarch64_gnullvm 0.42.2",
5072 + "windows_aarch64_msvc 0.42.2",
5073 + "windows_i686_gnu 0.42.2",
5074 + "windows_i686_msvc 0.42.2",
5075 + "windows_x86_64_gnu 0.42.2",
5076 + "windows_x86_64_gnullvm 0.42.2",
5077 + "windows_x86_64_msvc 0.42.2",
5078 +]
5079 +
5080 [[package]]
5081 name = "windows-targets"
5082 version = "0.52.6"
@@ -4802,6 +5110,12 @@ dependencies = [
5110 "windows_x86_64_msvc 0.53.1",
5111 ]
5112
5113 +[[package]]
5114 +name = "windows_aarch64_gnullvm"
5115 +version = "0.42.2"
5116 +source = "registry+https://github.com/rust-lang/crates.io-index"
5117 +checksum = "597a5118570b68bc08d8d59125332c54f1ba9d9adeedeef5b99b02ba2b0698f8"
5118 +
5119 [[package]]
5120 name = "windows_aarch64_gnullvm"
5121 version = "0.52.6"
@@ -4814,6 +5128,12 @@ version = "0.53.1"
5128 source = "registry+https://github.com/rust-lang/crates.io-index"
5129 checksum = "a9d8416fa8b42f5c947f8482c43e7d89e73a173cead56d044f6a56104a6d1b53"
5130
5131 +[[package]]
5132 +name = "windows_aarch64_msvc"
5133 +version = "0.42.2"
5134 +source = "registry+https://github.com/rust-lang/crates.io-index"
5135 +checksum = "e08e8864a60f06ef0d0ff4ba04124db8b0fb3be5776a5cd47641e942e58c4d43"
5136 +
5137 [[package]]
5138 name = "windows_aarch64_msvc"
5139 version = "0.52.6"
@@ -4826,6 +5146,12 @@ version = "0.53.1"
5146 source = "registry+https://github.com/rust-lang/crates.io-index"
5147 checksum = "b9d782e804c2f632e395708e99a94275910eb9100b2114651e04744e9b125006"
5148
5149 +[[package]]
5150 +name = "windows_i686_gnu"
5151 +version = "0.42.2"
5152 +source = "registry+https://github.com/rust-lang/crates.io-index"
5153 +checksum = "c61d927d8da41da96a81f029489353e68739737d3beca43145c8afec9a31a84f"
5154 +
5155 [[package]]
5156 name = "windows_i686_gnu"
5157 version = "0.52.6"
@@ -4850,6 +5176,12 @@ version = "0.53.1"
5176 source = "registry+https://github.com/rust-lang/crates.io-index"
5177 checksum = "fa7359d10048f68ab8b09fa71c3daccfb0e9b559aed648a8f95469c27057180c"
5178
5179 +[[package]]
5180 +name = "windows_i686_msvc"
5181 +version = "0.42.2"
5182 +source = "registry+https://github.com/rust-lang/crates.io-index"
5183 +checksum = "44d840b6ec649f480a41c8d80f9c65108b92d89345dd94027bfe06ac444d1060"
5184 +
5185 [[package]]
5186 name = "windows_i686_msvc"
5187 version = "0.52.6"
@@ -4862,6 +5194,12 @@ version = "0.53.1"
5194 source = "registry+https://github.com/rust-lang/crates.io-index"
5195 checksum = "1e7ac75179f18232fe9c285163565a57ef8d3c89254a30685b57d83a38d326c2"
5196
5197 +[[package]]
5198 +name = "windows_x86_64_gnu"
5199 +version = "0.42.2"
5200 +source = "registry+https://github.com/rust-lang/crates.io-index"
5201 +checksum = "8de912b8b8feb55c064867cf047dda097f92d51efad5b491dfb98f6bbb70cb36"
5202 +
5203 [[package]]
5204 name = "windows_x86_64_gnu"
5205 version = "0.52.6"
@@ -4874,6 +5212,12 @@ version = "0.53.1"
5212 source = "registry+https://github.com/rust-lang/crates.io-index"
5213 checksum = "9c3842cdd74a865a8066ab39c8a7a473c0778a3f29370b5fd6b4b9aa7df4a499"
5214
5215 +[[package]]
5216 +name = "windows_x86_64_gnullvm"
5217 +version = "0.42.2"
5218 +source = "registry+https://github.com/rust-lang/crates.io-index"
5219 +checksum = "26d41b46a36d453748aedef1486d5c7a85db22e56aff34643984ea85514e94a3"
5220 +
5221 [[package]]
5222 name = "windows_x86_64_gnullvm"
5223 version = "0.52.6"
@@ -4886,6 +5230,12 @@ version = "0.53.1"
5230 source = "registry+https://github.com/rust-lang/crates.io-index"
5231 checksum = "0ffa179e2d07eee8ad8f57493436566c7cc30ac536a3379fdf008f47f6bb7ae1"
5232
5233 +[[package]]
5234 +name = "windows_x86_64_msvc"
5235 +version = "0.42.2"
5236 +source = "registry+https://github.com/rust-lang/crates.io-index"
5237 +checksum = "9aec5da331524158c6d1a4ac0ab1541149c0b9505fde06423b02f5ef0106b9f0"
5238 +
5239 [[package]]
5240 name = "windows_x86_64_msvc"
5241 version = "0.52.6"
src/crates/Cargo.toml
+12 -14
@@ -63,10 +63,10 @@ lzma-rust2 = { version = "0.15", default-features = false, features = ["std", "x
63 md5 = "0.7"
64
65 roaring = { git = "https://github.com/netdata/roaring-rs.git", branch="allocative" }
66 -hashbrown = "0.15"
66 +hashbrown = "0.17.0"
67 serde = { version = "1.0" }
68 rustc-hash = "2.1"
69 -bincode = { version = "1.3" }
69 +rmp-serde = "1.3.1"
70 notify = "8.2"
71
72 walkdir = "2.4"
@@ -123,29 +123,27 @@ itoa = "1.0"
123 base64 = "0.22"
124
125 # NetFlow-specific dependencies
126 -jaq-core = "1.2.0"
127 -jaq-interpret = "1.2.0"
128 -jaq-std = "1.5.0"
129 -jaq-syn = "1.6.0"
130 -maxminddb = "0.25.0"
126 +jaq-core = "3.0.0"
127 +jaq-json = "2.0.0"
128 +jaq-std = "3.0.0"
129 +maxminddb = "0.27.3"
130 netflow_parser = "0.9.0"
132 -netgauze-bgp-pkt = "0.9.0"
133 -netgauze-bmp-pkt = "0.9.0"
131 +netgauze-bgp-pkt = "0.11.0"
132 +netgauze-bmp-pkt = "0.11.0"
133 sflow-parser = "0.6.0"
135 -reqwest = { version = "0.12.12", default-features = false }
136 -socket2 = "0.5.8"
134 +reqwest = { version = "0.13.2", default-features = false }
135 +socket2 = "0.6.3"
136 tonic-prost = "0.14"
137 prost = "0.14"
138 fst = "0.4.7"
139 tonic-prost-build = "0.14"
140 protoc-bin-vendored = "3"
142 -etherparse = "0.19.0"
141 +etherparse = "0.20.1"
142 pcap-file = "2.0.0"
143
144 # OTEL-specific dependencies
145 atty = "0.2"
147 -bytesize = "1.3"
148 -bytesize-serde = "0.2"
146 +bytesize = { version = "2.3.1", features = ["serde"] }
147 clap = { version = "4", features = ["derive"] }
148 duct = "0.13"
149 humantime = "2.2"
src/crates/netdata-log-viewer/otel-signal-viewer-plugin/Cargo.toml
-1
@@ -31,7 +31,6 @@ foyer = { workspace = true }
31 # Configuration support
32 anyhow = { workspace = true }
33 bytesize = { workspace = true }
34 -bytesize-serde = { workspace = true }
34 num_cpus = "1"
35 serde_yaml = { workspace = true }
36 parking_lot = { workspace = true }
src/crates/netdata-log-viewer/otel-signal-viewer-plugin/src/plugin_config.rs
-2
@@ -39,11 +39,9 @@ pub struct CacheConfig {
39 pub memory_capacity: usize,
40
41 /// Disk cache size (total size of disk-backed cache)
42 - #[serde(with = "bytesize_serde")]
42 pub disk_capacity: ByteSize,
43
44 /// Cache block size (size of cache blocks)
46 - #[serde(with = "bytesize_serde")]
45 pub block_size: ByteSize,
46
47 /// Number of background workers for indexing journal files
src/crates/netdata-otel/otel-plugin/Cargo.toml
-1
@@ -17,7 +17,6 @@ path = "src/main.rs"
17
18 [dependencies]
19 anyhow = { workspace = true }
20 -bytesize-serde = { workspace = true }
20 bytesize = { workspace = true }
21 clap = { workspace = true, features = ["derive"] }
22 humantime-serde = { workspace = true }
src/crates/netdata-otel/otel-plugin/src/plugin_config/logs.rs
-2
@@ -63,7 +63,6 @@ pub struct LogsConfig {
63 default_value = "100MB",
64 value_parser = parse_bytesize
65 )]
66 - #[serde(with = "bytesize_serde")]
66 pub size_of_journal_file: ByteSize,
67
68 /// Maximum number of entries in journal files
@@ -87,7 +86,6 @@ pub struct LogsConfig {
86 default_value = "1GB",
87 value_parser = parse_bytesize
88 )]
90 - #[serde(with = "bytesize_serde")]
89 pub size_of_journal_files: ByteSize,
90
91 /// Maximum age for journal entries (accepts human-readable durations like "7 days", "1 week", "168h")
src/crates/netflow-plugin/Cargo.toml
+4 -6
@@ -15,11 +15,9 @@ path = "src/main.rs"
15 anyhow = { workspace = true }
16 allocative = { workspace = true }
17 async-trait = { workspace = true }
18 -bincode = { workspace = true }
18 bitflags = { workspace = true }
19 bitvec = { workspace = true }
20 bytesize = { workspace = true }
22 -bytesize-serde = { workspace = true }
21 chrono = { workspace = true }
22 clap = { workspace = true, features = ["derive"] }
23 journal-common = { workspace = true }
@@ -36,10 +34,9 @@ ipnet = { workspace = true }
34 ipnet-trie = { workspace = true }
35 itoa = { workspace = true }
36 jaq-core = { workspace = true }
39 -jaq-interpret = { workspace = true }
37 +jaq-json = { workspace = true, features = ["serde"] }
38 jaq-std = { workspace = true }
41 -jaq-syn = { workspace = true }
42 -maxminddb = { workspace = true, features = ["mmap"] }
39 +maxminddb = { workspace = true }
40 memchr = { workspace = true }
41 memmap2 = { workspace = true }
42 netflow_parser = { workspace = true }
@@ -57,7 +54,7 @@ tokio-util = { workspace = true }
54 tracing = { workspace = true }
55 regex = { workspace = true }
56 twox-hash = { workspace = true, features = ["xxhash64"] }
60 -reqwest = { workspace = true, features = ["json", "rustls-tls"] }
57 +reqwest = { workspace = true, features = ["json", "rustls"] }
58 schemars = { workspace = true }
59 socket2 = { workspace = true }
60 tonic = { workspace = true, features = ["transport", "tls-native-roots"] }
@@ -66,6 +63,7 @@ prost = { workspace = true }
63 rustc-hash = { workspace = true }
64 thiserror = { workspace = true }
65 fst = { workspace = true }
66 +rmp-serde = { workspace = true }
67
68 rt = { workspace = true }
69 netdata-plugin-error = { workspace = true }
src/crates/netflow-plugin/src/api/flows/handler.rs
+3 -1
@@ -174,7 +174,9 @@ impl NetflowFlowsHandler {
174 }
175 }
176
177 -fn parse_flows_request(function_call: &FunctionCall) -> std::result::Result<query::FlowsRequest, FunctionResult> {
177 +fn parse_flows_request(
178 + function_call: &FunctionCall,
179 +) -> std::result::Result<query::FlowsRequest, FunctionResult> {
180 let request_value = if function_call.payload.is_some() {
181 payload_to_value(function_call)?
182 } else if function_call.args.is_empty() {
src/crates/netflow-plugin/src/decoder.rs
+1 -2
@@ -1,5 +1,4 @@
1 use crate::enrichment::FlowEnricher;
2 -use bincode::Options;
2 use netflow_parser::NetflowPacket;
3 use netflow_parser::scoped_parser::AutoScopedParser;
4 use netflow_parser::static_versions::{v5::V5, v7::V7};
@@ -85,7 +84,7 @@ const SFLOW_INTERFACE_LOCAL: u32 = 0x3fff_ffff;
84 const SFLOW_INTERFACE_FORMAT_INDEX: u32 = 0;
85 const SFLOW_INTERFACE_FORMAT_DISCARD: u32 = 1;
86 const VXLAN_UDP_PORT: u16 = 4789;
88 -const DECODER_STATE_SCHEMA_VERSION: u32 = 2;
87 +const DECODER_STATE_SCHEMA_VERSION: u32 = 3;
88 const DECODER_STATE_MAGIC: &[u8; 4] = b"NDFS";
89 const DECODER_STATE_HEADER_LEN: usize = 4 + 4 + 8 + 8;
90
src/crates/netflow-plugin/src/decoder/protocol/ipfix/record/state.rs
+4 -1
@@ -237,7 +237,10 @@ mod tests {
237 state.apply_reverse_time_overrides();
238
239 assert_eq!(
240 - state.reverse_overrides.get("FLOW_START_USEC").map(String::as_str),
240 + state
241 + .reverse_overrides
242 + .get("FLOW_START_USEC")
243 + .map(String::as_str),
244 Some("1042000")
245 );
246 }
src/crates/netflow-plugin/src/decoder/protocol/ipfix/special/record.rs
+1 -5
@@ -32,11 +32,7 @@ pub(crate) fn decode_ipfix_special_record(
32 let Some(value) = decode_akvorado_unsigned(raw_value) else {
33 continue;
34 };
35 - let status = if value & 0x03ff == 0 {
36 - "64"
37 - } else {
38 - "128"
39 - };
35 + let status = if value & 0x03ff == 0 { "64" } else { "128" };
36 fields.insert("FORWARDING_STATUS", status.to_string());
37 }
38 continue;
src/crates/netflow-plugin/src/decoder/state/persisted.rs
+10 -13
@@ -4,13 +4,6 @@ pub(crate) const MAX_DECODER_STATE_PAYLOAD_LEN: usize = 8 * 1024 * 1024;
4 pub(crate) const MAX_DECODER_STATE_FILE_LEN: usize =
5 DECODER_STATE_HEADER_LEN + MAX_DECODER_STATE_PAYLOAD_LEN;
6
7 -pub(crate) fn decoder_state_bincode_options() -> impl Options {
8 - bincode::DefaultOptions::new()
9 - .with_fixint_encoding()
10 - .with_little_endian()
11 - .with_limit(MAX_DECODER_STATE_PAYLOAD_LEN as u64)
12 -}
13 -
7 pub(crate) fn xxhash64(data: &[u8]) -> u64 {
8 let mut hasher = XxHash64::default();
9 hasher.write(data);
@@ -20,9 +13,15 @@ pub(crate) fn xxhash64(data: &[u8]) -> u64 {
13 pub(crate) fn encode_persisted_namespace_file(
14 file: &PersistedDecoderNamespaceFile,
15 ) -> Result<Vec<u8>, String> {
23 - let payload = decoder_state_bincode_options()
24 - .serialize(file)
16 + let payload = rmp_serde::to_vec_named(file)
17 .map_err(|err| format!("failed to encode decoder namespace state: {err}"))?;
18 + if payload.len() > MAX_DECODER_STATE_PAYLOAD_LEN {
19 + return Err(format!(
20 + "decoder namespace payload exceeds limit (max {} bytes, got {})",
21 + MAX_DECODER_STATE_PAYLOAD_LEN,
22 + payload.len()
23 + ));
24 + }
25 let payload_hash = xxhash64(&payload);
26 let payload_len = payload.len() as u64;
27
@@ -60,8 +59,7 @@ pub(crate) fn decode_persisted_namespace_file(
59 if payload_len > MAX_DECODER_STATE_PAYLOAD_LEN {
60 return Err(format!(
61 "decoder namespace payload exceeds limit (max {} bytes, got {})",
63 - MAX_DECODER_STATE_PAYLOAD_LEN,
64 - payload_len
62 + MAX_DECODER_STATE_PAYLOAD_LEN, payload_len
63 ));
64 }
65 let payload = &data[DECODER_STATE_HEADER_LEN..];
@@ -81,7 +79,6 @@ pub(crate) fn decode_persisted_namespace_file(
79 ));
80 }
81
84 - decoder_state_bincode_options()
85 - .deserialize(payload)
82 + rmp_serde::from_slice(payload)
83 .map_err(|err| format!("failed to decode decoder namespace state: {err}"))
84 }
src/crates/netflow-plugin/src/decoder/tests.rs
+4 -4
@@ -3,10 +3,10 @@ use super::{
3 DECODER_STATE_SCHEMA_VERSION, DIRECTION_EGRESS, DIRECTION_INGRESS, DecapsulationMode,
4 DecodeStats, DecodedFlow, DecoderStateNamespace, ETYPE_IPV4, ETYPE_IPV6, FlowDecoders,
5 FlowFields, FlowRecord, MAX_DECODER_STATE_PAYLOAD_LEN, SamplingState, TimestampSource,
6 - append_mpls_label, append_unique_flows, apply_icmp_port_fallback,
7 - apply_v9_special_mappings, decode_persisted_namespace_file, decode_v9_special_from_raw_payload,
8 - default_exporter_name, field_tracks_presence, finalize_canonical_flow_fields,
9 - normalize_direction_value, observe_v9_templates_from_raw_payload, to_field_token, xxhash64,
6 + append_mpls_label, append_unique_flows, apply_icmp_port_fallback, apply_v9_special_mappings,
7 + decode_persisted_namespace_file, decode_v9_special_from_raw_payload, default_exporter_name,
8 + field_tracks_presence, finalize_canonical_flow_fields, normalize_direction_value,
9 + observe_v9_templates_from_raw_payload, to_field_token, xxhash64,
10 };
11 use etherparse::{NetSlice, SlicedPacket, TransportSlice};
12 use netflow_parser::variable_versions::v9_lookup::V9Field;
src/crates/netflow-plugin/src/enrichment.rs
+1 -1
@@ -12,7 +12,7 @@ use crate::routing::DynamicRoutingRuntime;
12 use crate::routing::{DynamicRoutingPeerKey, DynamicRoutingUpdate};
13 use anyhow::{Context, Result};
14 use ipnet::IpNet;
15 -use maxminddb::{Mmap, Reader};
15 +use maxminddb::Reader;
16 use regex::Regex;
17 use serde::Deserialize;
18 use std::collections::{BTreeMap, HashMap};
src/crates/netflow-plugin/src/enrichment/data/geoip/files.rs
+1 -1
@@ -7,7 +7,7 @@ pub(crate) fn load_geoip_readers(
7 ) -> Result<Vec<GeoIpDatabaseReader>> {
8 let mut readers = Vec::new();
9 for path in paths {
10 - match Reader::open_mmap(path) {
10 + match Reader::open_readfile(path) {
11 Ok(reader) => readers.push(reader),
12 Err(err) if optional => {
13 tracing::warn!(
src/crates/netflow-plugin/src/enrichment/data/geoip/resolver.rs
+10 -18
@@ -1,12 +1,8 @@
1 use super::*;
2
3 -fn should_ignore_geoip_lookup_error(err: &maxminddb::MaxMindDBError) -> bool {
4 - matches!(err, maxminddb::MaxMindDBError::AddressNotFoundError(_))
5 -}
6 -
3 fn warn_unexpected_geoip_lookup_error(
4 database_kind: &'static str,
9 - err: &maxminddb::MaxMindDBError,
5 + err: &maxminddb::MaxMindDbError,
6 ) {
7 static ASN_LOOKUP_WARNED: std::sync::Once = std::sync::Once::new();
8 static GEO_LOOKUP_WARNED: std::sync::Once = std::sync::Once::new();
@@ -33,9 +29,15 @@ fn lookup_geoip_record<T>(
29 where
30 T: for<'de> serde::Deserialize<'de>,
31 {
36 - match db.lookup::<T>(address) {
37 - Ok(record) => Some(record),
38 - Err(err) if should_ignore_geoip_lookup_error(&err) => None,
32 + // Skip IPv6 lookups against IPv4-only databases; maxminddb would otherwise
33 + // return an InvalidInput error that we would have to detect by message.
34 + if address.is_ipv6() && db.metadata.ip_version == 4 {
35 + return None;
36 + }
37 +
38 + match db.lookup(address).and_then(|record| record.decode::<T>()) {
39 + Ok(Some(record)) => Some(record),
40 + Ok(None) => None,
41 Err(err) => {
42 warn_unexpected_geoip_lookup_error(database_kind, &err);
43 None
@@ -173,16 +175,6 @@ mod tests {
175 assert_eq!(resolver.last_reload_check, old_check);
176 }
177
176 - #[test]
177 - fn geoip_lookup_error_classifier_only_ignores_address_not_found() {
178 - assert!(should_ignore_geoip_lookup_error(
179 - &maxminddb::MaxMindDBError::AddressNotFoundError("missing".to_string())
180 - ));
181 - assert!(!should_ignore_geoip_lookup_error(
182 - &maxminddb::MaxMindDBError::InvalidDatabaseError("broken".to_string())
183 - ));
184 - }
185 -
178 #[test]
179 fn refresh_if_needed_keeps_retry_window_open_after_reader_reload_error() {
180 let dir = tempfile::tempdir().expect("create tempdir");
src/crates/netflow-plugin/src/enrichment/data/geoip/types.rs
+1 -1
@@ -1,6 +1,6 @@
1 use super::*;
2
3 -pub(crate) type GeoIpDatabaseReader = Reader<Mmap>;
3 +pub(crate) type GeoIpDatabaseReader = Reader<Vec<u8>>;
4
5 #[derive(Debug)]
6 pub(crate) struct GeoIpResolver {
src/crates/netflow-plugin/src/facet_runtime.rs
+135 -8
@@ -12,18 +12,19 @@ use crate::query::{
12 accumulate_targeted_facet_values, facet_field_requires_protocol_scan,
13 virtual_flow_field_dependencies,
14 };
15 -use anyhow::{Context, Result};
15 +use anyhow::{Context, Result, bail};
16 use journal_core::file::JournalFileMap;
17 use journal_registry::FileInfo;
18 use serde::{Deserialize, Serialize};
19 use std::collections::{BTreeMap, BTreeSet};
20 use std::fs;
21 -use std::io::BufReader;
21 +use std::hash::Hasher;
22 use std::mem::size_of;
23 use std::path::{Path, PathBuf};
24 use std::sync::atomic::{AtomicBool, Ordering};
25 use std::sync::{Arc, Mutex, RwLock};
26 use tokio::sync::Notify;
27 +use twox_hash::XxHash64;
28
29 #[allow(unused_imports)]
30 pub(crate) use contribution::{
@@ -33,8 +34,13 @@ pub(crate) use contribution::{
34 use sidecar::{delete_sidecar_files, search_sidecar, write_sidecar_files};
35 use store::{FacetStore, FacetStoreValueRef, PersistedFacetStore};
36
36 -const FACET_STATE_VERSION: u32 = 4;
37 +const FACET_STATE_VERSION: u32 = 5;
38 const FACET_STATE_FILE_NAME: &str = "facet-state.bin";
39 +const FACET_STATE_MAGIC: &[u8; 4] = b"NFFS";
40 +const FACET_STATE_SCHEMA_VERSION: u32 = 1;
41 +const FACET_STATE_HEADER_LEN: usize = 4 + 4 + 8 + 8;
42 +const MAX_FACET_STATE_PAYLOAD_LEN: usize = 128 * 1024 * 1024;
43 +const MAX_FACET_STATE_FILE_LEN: usize = FACET_STATE_HEADER_LEN + MAX_FACET_STATE_PAYLOAD_LEN;
44 const FACET_AUTOCOMPLETE_LIMIT: usize = 100;
45 const BTREE_ENTRY_OVERHEAD_BYTES: usize = size_of::<usize>() * 4;
46
@@ -815,7 +821,8 @@ fn persist_state_locked(state_path: &Path, state: &mut FacetState) -> Result<()>
821 .collect(),
822 published: state.published.fields.clone(),
823 };
818 - let payload = bincode::serialize(&persisted).context("failed to serialize facet state")?;
824 + let payload =
825 + encode_persisted_facet_state(&persisted).context("failed to serialize facet state")?;
826 let tmp_path = state_path.with_extension("bin.tmp");
827 fs::write(&tmp_path, &payload).with_context(|| {
828 format!(
@@ -834,6 +841,33 @@ fn persist_state_locked(state_path: &Path, state: &mut FacetState) -> Result<()>
841 Ok(())
842 }
843
844 +fn facet_state_xxhash64(data: &[u8]) -> u64 {
845 + let mut hasher = XxHash64::default();
846 + hasher.write(data);
847 + hasher.finish()
848 +}
849 +
850 +fn encode_persisted_facet_state(persisted: &PersistedFacetState) -> Result<Vec<u8>> {
851 + let payload = rmp_serde::to_vec_named(persisted)?;
852 + if payload.len() > MAX_FACET_STATE_PAYLOAD_LEN {
853 + bail!(
854 + "facet state payload exceeds limit (max {} bytes, got {})",
855 + MAX_FACET_STATE_PAYLOAD_LEN,
856 + payload.len()
857 + );
858 + }
859 +
860 + let payload_hash = facet_state_xxhash64(&payload);
861 + let payload_len = payload.len() as u64;
862 + let mut out = Vec::with_capacity(FACET_STATE_HEADER_LEN + payload.len());
863 + out.extend_from_slice(FACET_STATE_MAGIC);
864 + out.extend_from_slice(&FACET_STATE_SCHEMA_VERSION.to_le_bytes());
865 + out.extend_from_slice(&payload_hash.to_le_bytes());
866 + out.extend_from_slice(&payload_len.to_le_bytes());
867 + out.extend_from_slice(&payload);
868 + Ok(out)
869 +}
870 +
871 fn estimate_store_map_bytes(fields: &BTreeMap<String, FacetStore>) -> usize {
872 btree_container_overhead_bytes(fields.len())
873 + fields
@@ -928,8 +962,29 @@ fn btree_container_overhead_bytes(len: usize) -> usize {
962 }
963
964 fn load_persisted_state(state_path: &Path) -> Option<PersistedFacetState> {
931 - let file = match fs::File::open(state_path) {
932 - Ok(file) => file,
965 + let file_len = match fs::metadata(state_path) {
966 + Ok(metadata) => metadata.len(),
967 + Err(err) if err.kind() == std::io::ErrorKind::NotFound => return None,
968 + Err(err) => {
969 + tracing::warn!(
970 + "failed to stat persisted netflow facet state {}: {}",
971 + state_path.display(),
972 + err
973 + );
974 + return None;
975 + }
976 + };
977 + if file_len > MAX_FACET_STATE_FILE_LEN as u64 {
978 + tracing::warn!(
979 + "skipping oversized persisted netflow facet state {} (max {} bytes, got {})",
980 + state_path.display(),
981 + MAX_FACET_STATE_FILE_LEN,
982 + file_len
983 + );
984 + return None;
985 + }
986 + let data = match fs::read(state_path) {
987 + Ok(data) => data,
988 Err(err) if err.kind() == std::io::ErrorKind::NotFound => return None,
989 Err(err) => {
990 tracing::warn!(
@@ -940,8 +995,7 @@ fn load_persisted_state(state_path: &Path) -> Option<PersistedFacetState> {
995 return None;
996 }
997 };
943 - let persisted = match bincode::deserialize_from::<_, PersistedFacetState>(BufReader::new(file))
944 - {
998 + let persisted = match decode_persisted_facet_state(&data) {
999 Ok(persisted) => persisted,
1000 Err(err) => {
1001 tracing::warn!(
@@ -964,6 +1018,56 @@ fn load_persisted_state(state_path: &Path) -> Option<PersistedFacetState> {
1018 Some(persisted)
1019 }
1020
1021 +fn decode_persisted_facet_state(data: &[u8]) -> Result<PersistedFacetState> {
1022 + if data.len() < FACET_STATE_HEADER_LEN {
1023 + bail!("truncated netflow facet state header");
1024 + }
1025 + if &data[..4] != FACET_STATE_MAGIC {
1026 + bail!("invalid netflow facet state magic");
1027 + }
1028 +
1029 + let version = u32::from_le_bytes(data[4..8].try_into().unwrap());
1030 + if version != FACET_STATE_SCHEMA_VERSION {
1031 + bail!(
1032 + "unsupported netflow facet state schema version {} (expected {})",
1033 + version,
1034 + FACET_STATE_SCHEMA_VERSION
1035 + );
1036 + }
1037 +
1038 + let expected_hash = u64::from_le_bytes(data[8..16].try_into().unwrap());
1039 + let payload_len = u64::from_le_bytes(data[16..24].try_into().unwrap());
1040 + let payload_len = usize::try_from(payload_len)
1041 + .context("netflow facet state payload length overflows usize")?;
1042 + if payload_len > MAX_FACET_STATE_PAYLOAD_LEN {
1043 + bail!(
1044 + "netflow facet state payload exceeds limit (max {} bytes, got {})",
1045 + MAX_FACET_STATE_PAYLOAD_LEN,
1046 + payload_len
1047 + );
1048 + }
1049 +
1050 + let payload = &data[FACET_STATE_HEADER_LEN..];
1051 + if payload.len() != payload_len {
1052 + bail!(
1053 + "netflow facet state payload length mismatch (header {}, actual {})",
1054 + payload_len,
1055 + payload.len()
1056 + );
1057 + }
1058 +
1059 + let actual_hash = facet_state_xxhash64(payload);
1060 + if actual_hash != expected_hash {
1061 + bail!(
1062 + "netflow facet state payload hash mismatch (expected {}, got {})",
1063 + expected_hash,
1064 + actual_hash
1065 + );
1066 + }
1067 +
1068 + rmp_serde::from_slice(payload).context("failed to decode netflow facet state payload")
1069 +}
1070 +
1071 fn merge_autocomplete_values(left: Vec<String>, right: Vec<String>) -> Vec<String> {
1072 let mut merged = left.into_iter().collect::<BTreeSet<_>>();
1073 merged.extend(right);
@@ -1153,6 +1257,29 @@ mod tests {
1257 }
1258 }
1259
1260 + #[test]
1261 + fn load_persisted_state_skips_oversized_file_without_full_read() {
1262 + let tmp = tempfile::tempdir().expect("create temp dir");
1263 + let state_path = tmp.path().join(FACET_STATE_FILE_NAME);
1264 + // Use a sparse file so we exceed the cap without actually writing 128MiB.
1265 + let oversized = (MAX_FACET_STATE_FILE_LEN as u64).saturating_add(1);
1266 + let file = std::fs::File::create(&state_path).expect("create state file");
1267 + file.set_len(oversized).expect("extend state file");
1268 + drop(file);
1269 +
1270 + let metadata = std::fs::metadata(&state_path).expect("stat state file");
1271 + assert!(
1272 + metadata.len() > MAX_FACET_STATE_FILE_LEN as u64,
1273 + "test setup must produce an oversized file"
1274 + );
1275 +
1276 + let loaded = load_persisted_state(&state_path);
1277 + assert!(
1278 + loaded.is_none(),
1279 + "oversized facet state file must be skipped, not loaded"
1280 + );
1281 + }
1282 +
1283 #[test]
1284 #[ignore = "manual production-data facet allocative profiler"]
1285 fn stress_profile_live_archived_facet_allocative_breakdown() {
src/crates/netflow-plugin/src/facet_runtime/contribution.rs
-1
@@ -560,5 +560,4 @@ mod tests {
560 contribution_strings(&encoded)
561 );
562 }
563 -
563 }
src/crates/netflow-plugin/src/ingest.rs
+3 -3
@@ -65,9 +65,6 @@ fn tier_timestamp_lookup_query_end(now_usec: u64) -> u32 {
65 .max(1) as u32
66 }
67
68 -#[cfg(test)]
69 -#[path = "ingest_test_support.rs"]
70 -mod test_support;
68 #[cfg(test)]
69 #[path = "ingest_bench_support.rs"]
70 mod bench_support;
@@ -81,5 +78,8 @@ mod resource_bench_support;
78 #[path = "ingest_resource_bench_tests.rs"]
79 mod resource_bench_tests;
80 #[cfg(test)]
81 +#[path = "ingest_test_support.rs"]
82 +mod test_support;
83 +#[cfg(test)]
84 #[path = "ingest_tests.rs"]
85 mod tests;
src/crates/netflow-plugin/src/ingest/encode.rs
+4 -1
@@ -104,7 +104,10 @@ impl JournalEncodeBuffer {
104
105 #[cfg(test)]
106 pub(crate) fn debug_field_slices(&self) -> Vec<&[u8]> {
107 - self.refs.iter().map(|range| &self.data[range.clone()]).collect()
107 + self.refs
108 + .iter()
109 + .map(|range| &self.data[range.clone()])
110 + .collect()
111 }
112
113 fn push_number(&mut self, name: &str, value: u64) {
src/crates/netflow-plugin/src/ingest/service/tiers.rs
+5 -1
@@ -109,7 +109,11 @@ impl IngestService {
109 .facet_runtime
110 .observe_active_contribution(Path::new(active_file.path()), &contribution)
111 {
112 - tracing::warn!("facet runtime tier {:?} write update failed: {}", tier, err);
112 + tracing::warn!(
113 + "facet runtime tier {:?} write update failed: {}",
114 + tier,
115 + err
116 + );
117 }
118 }
119 self.metrics
src/crates/netflow-plugin/src/ingest_bench_support.rs
+3 -1
@@ -101,7 +101,9 @@ pub(super) fn warm_protocol_templates(service: &mut IngestService, scenario: &Pr
101 let base = fixture_dir();
102 for file in scenario.template_files {
103 for payload in extract_udp_payloads(&base.join(file)) {
104 - service.decoders.decode_udp_payload(payload.source, &payload.data);
104 + service
105 + .decoders
106 + .decode_udp_payload(payload.source, &payload.data);
107 }
108 }
109 }
src/crates/netflow-plugin/src/ingest_bench_tests.rs
+9 -8
@@ -1,9 +1,9 @@
1 -use super::test_support::{UdpPayload, new_benchmark_ingest_service};
1 use super::bench_support::{
2 CARDINALITY_SOURCE_SCENARIO, CardinalityMode, PROTOCOL_SCENARIOS, ProtocolScenario,
3 build_cardinality_records, collect_decoded_flows, collect_decoded_flows_for_scenario,
4 count_flows_per_round, load_scenario_payloads, total_payload_bytes, warm_protocol_templates,
5 };
6 +use super::test_support::{UdpPayload, new_benchmark_ingest_service};
7 use crate::decoder::DecodedFlow;
8 use crate::plugin_config::DecapsulationMode as ConfigDecapsulationMode;
9 use std::time::Instant;
@@ -188,9 +188,11 @@ fn benchmark_protocol_decode_only(
188 for _ in 0..rounds {
189 for payload in data_payloads {
190 let receive_time_usec = super::now_usec();
191 - let batch = service
192 - .decoders
193 - .decode_udp_payload_at(payload.source, &payload.data, receive_time_usec);
191 + let batch = service.decoders.decode_udp_payload_at(
192 + payload.source,
193 + &payload.data,
194 + receive_time_usec,
195 + );
196 flows += batch.flows.len();
197 }
198 }
@@ -216,10 +218,9 @@ fn benchmark_protocol_post_decode(
218
219 for _ in 0..rounds {
220 for flow in &decoded {
219 - if service.ingest_decoded_record_for_test(
220 - CARDINALITY_BENCH_RECEIVE_TIME_USEC,
221 - &flow.record,
222 - ) {
221 + if service
222 + .ingest_decoded_record_for_test(CARDINALITY_BENCH_RECEIVE_TIME_USEC, &flow.record)
223 + {
224 entries_since_sync += 1;
225 }
226 }
src/crates/netflow-plugin/src/ingest_resource_bench_support.rs
+4 -1
@@ -116,7 +116,10 @@ pub(super) fn parse_child_report(output: &std::process::Output) -> ResourceEnvel
116 let combined = format!("{stdout}\n{stderr}");
117 let json = combined
118 .lines()
119 - .find_map(|line| line.split_once("RESOURCE_BENCH_RESULT:").map(|(_, json)| json))
119 + .find_map(|line| {
120 + line.split_once("RESOURCE_BENCH_RESULT:")
121 + .map(|(_, json)| json)
122 + })
123 .unwrap_or_else(|| panic!("resource bench child did not emit result\n{combined}"));
124 serde_json::from_str(json)
125 .unwrap_or_else(|err| panic!("parse resource bench result JSON: {err}\n{combined}"))
src/crates/netflow-plugin/src/ingest_resource_bench_tests.rs
+41 -26
@@ -177,7 +177,10 @@ fn run_resource_envelope_case(
177 },
178 )
179 .env(RATE_ENV, flows_per_sec.to_string())
180 - .env(WARMUP_ENV, env_u64(WARMUP_ENV, DEFAULT_WARMUP_SECS).to_string())
180 + .env(
181 + WARMUP_ENV,
182 + env_u64(WARMUP_ENV, DEFAULT_WARMUP_SECS).to_string(),
183 + )
184 .env(
185 MEASURE_ENV,
186 env_u64(MEASURE_ENV, DEFAULT_MEASURE_SECS).to_string(),
@@ -211,12 +214,9 @@ fn run_resource_envelope_child() -> ResourceEnvelopeReport {
214 let warmup_secs = env_u64(WARMUP_ENV, DEFAULT_WARMUP_SECS);
215 let measurement_secs = env_u64(MEASURE_ENV, DEFAULT_MEASURE_SECS);
216 match layer {
214 - ResourceLayer::WriterOnly => run_writer_only_resource_envelope(
215 - profile,
216 - flows_per_sec,
217 - warmup_secs,
218 - measurement_secs,
219 - ),
217 + ResourceLayer::WriterOnly => {
218 + run_writer_only_resource_envelope(profile, flows_per_sec, warmup_secs, measurement_secs)
219 + }
220 ResourceLayer::RawOnly | ResourceLayer::Minute1Only | ResourceLayer::AllTiersBatched => {
221 run_plugin_resource_envelope(
222 layer,
@@ -270,8 +270,8 @@ fn run_writer_only_resource_envelope(
270 log.sync().expect("sync isolated writer benchmark log");
271
272 ResourceEnvelopeReport {
273 - methodology:
274 - "paced mixed-flow raw journal benchmark with a single disk-backed writer".to_string(),
273 + methodology: "paced mixed-flow raw journal benchmark with a single disk-backed writer"
274 + .to_string(),
275 layer: ResourceLayer::WriterOnly.label().to_string(),
276 profile: profile.label().to_string(),
277 requested_flows_per_sec: flows_per_sec,
@@ -281,9 +281,7 @@ fn run_writer_only_resource_envelope(
281 / elapsed.as_secs_f64(),
282 logical_entries_per_sec: measurement_result.logical_entries_written as f64
283 / elapsed.as_secs_f64(),
284 - read_bytes_per_sec: proc_after
285 - .read_bytes
286 - .saturating_sub(proc_before.read_bytes) as f64
284 + read_bytes_per_sec: proc_after.read_bytes.saturating_sub(proc_before.read_bytes) as f64
285 / elapsed.as_secs_f64(),
286 write_bytes_per_sec: proc_after
287 .write_bytes
@@ -340,14 +338,26 @@ fn run_plugin_resource_envelope(
338 service.finish_shutdown_for_test(measurement_result.entries_since_sync);
339
340 let logical_bytes = match layer {
343 - ResourceLayer::RawOnly => counter_delta(&metrics_before, &metrics_after, "raw_journal_logical_bytes"),
344 - ResourceLayer::Minute1Only => counter_delta(&metrics_before, &metrics_after, "raw_journal_logical_bytes")
345 - .saturating_add(counter_delta(&metrics_before, &metrics_after, "minute_1_logical_bytes")),
346 - ResourceLayer::AllTiersBatched => total_logical_bytes_delta(&metrics_before, &metrics_after),
341 + ResourceLayer::RawOnly => {
342 + counter_delta(&metrics_before, &metrics_after, "raw_journal_logical_bytes")
343 + }
344 + ResourceLayer::Minute1Only => {
345 + counter_delta(&metrics_before, &metrics_after, "raw_journal_logical_bytes")
346 + .saturating_add(counter_delta(
347 + &metrics_before,
348 + &metrics_after,
349 + "minute_1_logical_bytes",
350 + ))
351 + }
352 + ResourceLayer::AllTiersBatched => {
353 + total_logical_bytes_delta(&metrics_before, &metrics_after)
354 + }
355 ResourceLayer::WriterOnly => 0,
356 };
357 let entries_written = match layer {
350 - ResourceLayer::RawOnly => counter_delta(&metrics_before, &metrics_after, "journal_entries_written"),
358 + ResourceLayer::RawOnly => {
359 + counter_delta(&metrics_before, &metrics_after, "journal_entries_written")
360 + }
361 ResourceLayer::Minute1Only | ResourceLayer::AllTiersBatched => {
362 total_entries_written_delta(&metrics_before, &metrics_after)
363 }
@@ -355,8 +365,8 @@ fn run_plugin_resource_envelope(
365 };
366
367 ResourceEnvelopeReport {
358 - methodology:
359 - "post-decode paced mixed-flow ingest benchmark with disk-backed journals".to_string(),
368 + methodology: "post-decode paced mixed-flow ingest benchmark with disk-backed journals"
369 + .to_string(),
370 layer: layer.label().to_string(),
371 profile: profile.label().to_string(),
372 requested_flows_per_sec: flows_per_sec,
@@ -364,9 +374,7 @@ fn run_plugin_resource_envelope(
374 cpu_percent_of_one_core: cpu_percent_of_one_core(proc_before, proc_after, elapsed),
375 logical_write_bytes_per_sec: logical_bytes as f64 / elapsed.as_secs_f64(),
376 logical_entries_per_sec: entries_written as f64 / elapsed.as_secs_f64(),
367 - read_bytes_per_sec: proc_after
368 - .read_bytes
369 - .saturating_sub(proc_before.read_bytes) as f64
377 + read_bytes_per_sec: proc_after.read_bytes.saturating_sub(proc_before.read_bytes) as f64
378 / elapsed.as_secs_f64(),
379 write_bytes_per_sec: proc_after
380 .write_bytes
@@ -388,8 +396,12 @@ fn run_plugin_resource_envelope(
396
397 fn configure_service_for_layer(service: &mut IngestService, layer: ResourceLayer) {
398 if matches!(layer, ResourceLayer::Minute1Only) {
391 - service.tier_accumulators.remove(&crate::tiering::TierKind::Minute5);
392 - service.tier_accumulators.remove(&crate::tiering::TierKind::Hour1);
399 + service
400 + .tier_accumulators
401 + .remove(&crate::tiering::TierKind::Minute5);
402 + service
403 + .tier_accumulators
404 + .remove(&crate::tiering::TierKind::Hour1);
405 }
406 }
407
@@ -561,8 +573,11 @@ fn total_entries_written_delta(
573 before: &std::collections::HashMap<String, u64>,
574 after: &std::collections::HashMap<String, u64>,
575 ) -> u64 {
564 - counter_delta(before, after, "journal_entries_written")
565 - .saturating_add(counter_delta(before, after, "tier_entries_written"))
576 + counter_delta(before, after, "journal_entries_written").saturating_add(counter_delta(
577 + before,
578 + after,
579 + "tier_entries_written",
580 + ))
581 }
582
583 fn total_logical_bytes_delta(
src/crates/netflow-plugin/src/ingest_test_support.rs
+8 -2
@@ -88,7 +88,8 @@ pub(super) fn new_disk_benchmark_raw_log() -> (TempDir, Log) {
88 let retention = cfg.journal.retention_for_tier(TierKind::Raw);
89 let mut retention_policy = RetentionPolicy::default();
90 if let Some(size_of_journal_files) = retention.size_of_journal_files {
91 - retention_policy = retention_policy.with_size_of_journal_files(size_of_journal_files.as_u64());
91 + retention_policy =
92 + retention_policy.with_size_of_journal_files(size_of_journal_files.as_u64());
93 }
94 if let Some(duration_of_journal_files) = retention.duration_of_journal_files {
95 retention_policy =
@@ -153,7 +154,12 @@ fn decode_pcap_flows(path: &Path, service: &mut IngestService) -> Vec<crate::dec
154 if let Some((source, payload)) = extract_udp_payload(packet.data.as_ref()) {
155 service.prepare_decoder_state_namespace(source, payload);
156 let decoded = service.decoders.decode_udp_payload(source, payload);
156 - flows.extend(decoded.flows.into_iter().map(|flow| flow.record.to_fields()));
157 + flows.extend(
158 + decoded
159 + .flows
160 + .into_iter()
161 + .map(|flow| flow.record.to_fields()),
162 + );
163 }
164 }
165 flows
src/crates/netflow-plugin/src/ingest_tests.rs
+1 -1
@@ -1,7 +1,7 @@
1 -use super::*;
1 use super::test_support::{
2 decode_fixture_sequence, find_flow, new_test_ingest_service, new_test_ingest_service_in_dir,
3 };
4 +use super::*;
5 use crate::plugin_config::DecapsulationMode as ConfigDecapsulationMode;
6
7 #[test]
src/crates/netflow-plugin/src/main.rs
+2 -2
@@ -5,10 +5,10 @@ mod charts;
5 mod decoder;
6 mod enrichment;
7 mod facet_catalog;
8 -#[allow(dead_code)]
9 -mod flow_index;
8 mod facet_runtime;
9 mod flow;
10 +#[allow(dead_code)]
11 +mod flow_index;
12 mod ingest;
13 mod memory_allocator;
14 #[cfg(test)]
src/crates/netflow-plugin/src/network_sources/mod.rs
+1 -1
@@ -9,12 +9,12 @@ mod types;
9
10 pub(crate) use runtime::{NetworkSourceRecord, NetworkSourcesRuntime};
11 pub(crate) use service::run_network_sources_refresher;
12 +pub(crate) use transform::compile_jaq_filter;
13
14 use crate::enrichment::NetworkAttributes;
15 use crate::plugin_config::{RemoteNetworkSourceConfig, RemoteNetworkSourceTlsConfig};
16 use anyhow::{Context, Result};
17 use ipnet::IpNet;
17 -use jaq_interpret::{Ctx, Filter, FilterT, ParseCtx, RcIter, Val};
18 use reqwest::{Certificate, Client, Identity, Method};
19 use serde::Deserialize;
20 use serde_json::Value;
src/crates/netflow-plugin/src/network_sources/transform.rs
+122 -25
@@ -1,5 +1,8 @@
1 use super::types::CompiledTransform;
2 use super::*;
3 +use jaq_core::load::{Arena, File, Loader};
4 +use jaq_core::{Compiler, Ctx, Vars, compile, data, load, unwrap_valr};
5 +use jaq_json::Val as JaqVal;
6
7 pub(super) fn compile_transform(expression: &str) -> Result<CompiledTransform> {
8 let normalized = if expression.trim().is_empty() {
@@ -8,27 +11,7 @@ pub(super) fn compile_transform(expression: &str) -> Result<CompiledTransform> {
11 expression.trim().to_string()
12 };
13
11 - let tokens = jaq_syn::Lexer::new(&normalized)
12 - .lex()
13 - .map_err(|errs| anyhow::anyhow!("failed to lex transform '{}': {:?}", normalized, errs))?;
14 - let main = jaq_syn::Parser::new(&tokens)
15 - .parse(|parser| parser.module(|module| module.term()))
16 - .map_err(|errs| anyhow::anyhow!("failed to parse transform '{}': {:?}", normalized, errs))?
17 - .conv(&normalized);
18 -
19 - let mut ctx = ParseCtx::new(Vec::new());
20 - ctx.insert_natives(jaq_core::core());
21 - ctx.insert_defs(jaq_std::std());
22 - let filter = ctx.compile(main);
23 - if !ctx.errs.is_empty() {
24 - let errors = ctx
25 - .errs
26 - .into_iter()
27 - .map(|err| err.0.to_string())
28 - .collect::<Vec<_>>()
29 - .join("; ");
30 - anyhow::bail!("failed to compile transform '{}': {}", normalized, errors);
31 - }
14 + let filter = compile_jaq_filter(&normalized)?;
15
16 Ok(CompiledTransform {
17 expression: normalized,
@@ -37,9 +20,10 @@ pub(super) fn compile_transform(expression: &str) -> Result<CompiledTransform> {
20 }
21
22 pub(super) fn run_transform(payload: Value, transform: &CompiledTransform) -> Result<Vec<Value>> {
40 - let input = Val::from(payload);
41 - let inputs = RcIter::new(core::iter::empty());
42 - let mut output = transform.filter.run((Ctx::new([], &inputs), input));
23 + let input: JaqVal =
24 + serde_json::from_value(payload).context("failed to convert JSON payload to jaq value")?;
25 + let ctx = Ctx::<data::JustLut<JaqVal>>::new(&transform.filter.lut, Vars::new([]));
26 + let mut output = transform.filter.id.run((ctx, input)).map(unwrap_valr);
27 let mut rows = Vec::new();
28 while let Some(next) = output.next() {
29 let value = next.map_err(|err| {
@@ -49,10 +33,123 @@ pub(super) fn run_transform(payload: Value, transform: &CompiledTransform) -> Re
33 err
34 )
35 })?;
52 - rows.push(Value::from(value));
36 + rows.push(jaq_value_to_json(value).with_context(|| {
37 + format!(
38 + "failed to convert transform '{}' result to JSON",
39 + transform.expression
40 + )
41 + })?);
42 }
43 if rows.is_empty() {
44 anyhow::bail!("transform '{}' produced empty result", transform.expression);
45 }
46 Ok(rows)
47 }
48 +
49 +pub(crate) fn compile_jaq_filter(
50 + expression: &str,
51 +) -> Result<jaq_core::Filter<jaq_core::data::JustLut<JaqVal>>> {
52 + let defs = jaq_core::defs()
53 + .chain(jaq_std::defs())
54 + .chain(jaq_json::defs());
55 + let funs = jaq_core::funs()
56 + .chain(jaq_std::funs())
57 + .chain(jaq_json::funs());
58 + let loader = Loader::new(defs);
59 + let arena = Arena::default();
60 + let modules = loader
61 + .load(
62 + &arena,
63 + File {
64 + code: expression,
65 + path: (),
66 + },
67 + )
68 + .map_err(|errs| {
69 + anyhow::anyhow!(
70 + "failed to parse transform '{}': {}",
71 + expression,
72 + format_load_errors(&errs)
73 + )
74 + })?;
75 +
76 + Compiler::<_, data::JustLut<JaqVal>>::default()
77 + .with_funs(funs)
78 + .compile(modules)
79 + .map_err(|errs| {
80 + anyhow::anyhow!(
81 + "failed to compile transform '{}': {}",
82 + expression,
83 + format_compile_errors(&errs)
84 + )
85 + })
86 +}
87 +
88 +fn format_load_errors<P>(errs: &load::Errors<&str, P>) -> String {
89 + let mut parts = Vec::new();
90 + for (_file, err) in errs {
91 + match err {
92 + load::Error::Io(items) => {
93 + for (_, msg) in items {
94 + parts.push(format!("io error: {msg}"));
95 + }
96 + }
97 + load::Error::Lex(items) => {
98 + for (expected, found) in items {
99 + parts.push(format!(
100 + "lex error near \"{}\": expected {:?}",
101 + truncate_for_display(found),
102 + expected
103 + ));
104 + }
105 + }
106 + load::Error::Parse(items) => {
107 + for (expected, input) in items {
108 + parts.push(format!(
109 + "parse error near \"{}\": expected {:?}",
110 + truncate_for_display(input),
111 + expected
112 + ));
113 + }
114 + }
115 + }
116 + }
117 + if parts.is_empty() {
118 + "unknown error".to_string()
119 + } else {
120 + parts.join("; ")
121 + }
122 +}
123 +
124 +fn format_compile_errors<P>(errs: &compile::Errors<&str, P>) -> String {
125 + let mut parts = Vec::new();
126 + for (_file, items) in errs {
127 + for (sym, undefined) in items {
128 + parts.push(format!(
129 + "undefined {} \"{}\"",
130 + undefined.as_str(),
131 + truncate_for_display(sym)
132 + ));
133 + }
134 + }
135 + if parts.is_empty() {
136 + "unknown error".to_string()
137 + } else {
138 + parts.join("; ")
139 + }
140 +}
141 +
142 +fn truncate_for_display(s: &str) -> String {
143 + const MAX: usize = 60;
144 + let trimmed = s.trim();
145 + if trimmed.chars().count() <= MAX {
146 + trimmed.to_string()
147 + } else {
148 + let cut: String = trimmed.chars().take(MAX).collect();
149 + format!("{cut}...")
150 + }
151 +}
152 +
153 +fn jaq_value_to_json(value: JaqVal) -> Result<Value> {
154 + serde_json::from_str(&value.to_string()).context("jaq value is not representable as JSON")
155 +}
src/crates/netflow-plugin/src/network_sources/types.rs
+9 -2
@@ -39,8 +39,15 @@ pub(super) struct SourceRecordState {
39 pub(super) by_source: Arc<RwLock<BTreeMap<String, Vec<NetworkSourceRecord>>>>,
40 }
41
42 -#[derive(Debug, Clone)]
42 pub(super) struct CompiledTransform {
43 pub(super) expression: String,
45 - pub(super) filter: Filter,
44 + pub(super) filter: jaq_core::Filter<jaq_core::data::JustLut<jaq_json::Val>>,
45 +}
46 +
47 +impl std::fmt::Debug for CompiledTransform {
48 + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
49 + f.debug_struct("CompiledTransform")
50 + .field("expression", &self.expression)
51 + .finish_non_exhaustive()
52 + }
53 }
src/crates/netflow-plugin/src/plugin_config/validation/enrichment.rs
+4 -29
@@ -1,6 +1,6 @@
1 +use crate::network_sources::compile_jaq_filter;
2 use crate::plugin_config::{NetworkAttributesValue, PluginConfig, RemoteNetworkSourceTlsConfig};
3 use anyhow::{Context, Result};
3 -use jaq_interpret::ParseCtx;
4 use std::net::SocketAddr;
5 use std::time::Duration;
6
@@ -154,37 +154,12 @@ fn validate_network_source_transform(source_name: &str, transform: &str) -> Resu
154 } else {
155 transform.trim()
156 };
157 - let tokens = jaq_syn::Lexer::new(normalized).lex().map_err(|errs| {
157 + compile_jaq_filter(normalized).map_err(|err| {
158 anyhow::anyhow!(
159 - "enrichment.network_sources.{source_name}.transform lex error: {:?}",
160 - errs
159 + "enrichment.network_sources.{source_name}.transform compile error: {}",
160 + err
161 )
162 })?;
163 - let main = jaq_syn::Parser::new(&tokens)
164 - .parse(|parser| parser.module(|module| module.term()))
165 - .map_err(|errs| {
166 - anyhow::anyhow!(
167 - "enrichment.network_sources.{source_name}.transform parse error: {:?}",
168 - errs
169 - )
170 - })?
171 - .conv(normalized);
172 - let mut ctx = ParseCtx::new(Vec::new());
173 - ctx.insert_natives(jaq_core::core());
174 - ctx.insert_defs(jaq_std::std());
175 - let _compiled = ctx.compile(main);
176 - if !ctx.errs.is_empty() {
177 - let errors = ctx
178 - .errs
179 - .into_iter()
180 - .map(|err| err.0.to_string())
181 - .collect::<Vec<_>>()
182 - .join("; ");
183 - anyhow::bail!(
184 - "enrichment.network_sources.{source_name}.transform compile error: {}",
185 - errors
186 - );
187 - }
163 Ok(())
164 }
165
src/crates/netflow-plugin/src/query.rs
+4 -4
@@ -1,4 +1,8 @@
1 use crate::flow::canonical_flow_field_names;
2 +use crate::flow_index::{
3 + FieldKind as IndexFieldKind, FieldSpec as IndexFieldSpec, FieldValue as IndexFieldValue,
4 + FlowId as IndexedFlowId, FlowIndex,
5 +};
6 use crate::plugin_config::PluginConfig;
7 use crate::presentation;
8 use crate::tiering::TierKind;
@@ -13,10 +17,6 @@ use journal_core::{
17 Direction as JournalDirection, JournalCursor, JournalFile, JournalReader, Location,
18 };
19 use journal_registry::{FileInfo, Monitor, Registry, repository::File as RegistryFile};
16 -use crate::flow_index::{
17 - FieldKind as IndexFieldKind, FieldSpec as IndexFieldSpec, FieldValue as IndexFieldValue,
18 - FlowId as IndexedFlowId, FlowIndex,
19 -};
20 use notify::Event;
21 use regex::Regex;
22 use serde::de::Error as _;
src/crates/netflow-plugin/src/tiering/index/store.rs
+3 -3
@@ -10,12 +10,12 @@ use super::super::rollup::{
10 };
11 use crate::facet_runtime::FacetFileContribution;
12 use crate::flow::{FlowFields, FlowRecord};
13 -use crate::ingest::JournalEncodeBuffer;
14 -use crate::tiering::FlowMetrics;
15 -use anyhow::{Context, Result, anyhow};
13 #[cfg(test)]
14 use crate::flow_index::FieldValue as IndexFieldValue;
15 use crate::flow_index::{FlowIndex, FlowIndexMemoryBreakdown};
16 +use crate::ingest::JournalEncodeBuffer;
17 +use crate::tiering::FlowMetrics;
18 +use anyhow::{Context, Result, anyhow};
19 use std::collections::{BTreeMap, BTreeSet};
20 use std::mem::size_of;
21
src/crates/netflow-plugin/src/tiering/rollup.rs
+2 -2
@@ -5,14 +5,14 @@ use crate::flow_index::{
5 };
6 use std::net::{IpAddr, Ipv4Addr};
7
8 -mod encode;
8 mod emit;
9 +mod encode;
10 mod materialize;
11 mod schema;
12
13 +pub(crate) use emit::emit_rollup_row;
14 #[allow(unused_imports)]
15 pub(crate) use encode::push_rollup_field_ids;
15 -pub(crate) use emit::emit_rollup_row;
16 #[cfg(test)]
17 pub(crate) use materialize::dimensions_for_rollup;
18 #[allow(unused_imports)]