diff --git a/Cargo.lock b/Cargo.lock index c44c291..452bdfb 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -60,6 +60,15 @@ version = "2.13.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3ded4057c258ba199e2d26386d3af3780957ecaee6c4ef4041c6b4b8b97c0b06" +[[package]] +name = "block-buffer" +version = "0.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d2f6c7dbe95a6ed67ad9f18e57daf93a2f034c524b99fd2b76d18fdfeb6660aa" +dependencies = [ + "hybrid-array", +] + [[package]] name = "bumpalo" version = "3.20.3" @@ -123,6 +132,12 @@ dependencies = [ "cc", ] +[[package]] +name = "cmov" +version = "0.5.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0c9ea0ac24bc397ab3c98583a3c9ba74fa56b09a4449bbe172b9b1ddb016027a" + [[package]] name = "config" version = "0.15.27" @@ -135,6 +150,21 @@ dependencies = [ "winnow", ] +[[package]] +name = "const-oid" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a6ef517f0926dd24a1582492c791b6a4818a4d94e789a334894aa15b0d12f55c" + +[[package]] +name = "crypto-common" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ce6e4c961d6cd6c9a86db418387425e8bdeaf05b3c8bc1411e6dca4c252f1453" +dependencies = [ + "hybrid-array", +] + [[package]] name = "csv" version = "1.4.0" @@ -156,6 +186,27 @@ dependencies = [ "memchr", ] +[[package]] +name = "ctutils" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "03bb0e1cc970d482d121d9a1744999169b69a07470b3d644a7894e53fcaf4574" +dependencies = [ + "cmov", +] + +[[package]] +name = "digest" +version = "0.11.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f1dd6dbb5841937940781866fa1281a1ff7bd3bf827091440879f9994983d5c2" +dependencies = [ + "block-buffer", + "const-oid", + "crypto-common", + "ctutils", +] + [[package]] name = "either" version = "1.19.0" @@ -242,6 +293,24 @@ version = "0.17.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a" +[[package]] +name = "hmac" +version = "0.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6303bc9732ae41b04cb554b844a762b4115a61bfaa81e3e83050991eeb56863f" +dependencies = [ + "digest", +] + +[[package]] +name = "hybrid-array" +version = "0.4.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "27f864f10dfb56725ce5ce5472bc52252c8f93a4ab86327122cebf62c5f59a17" +dependencies = [ + "typenum", +] + [[package]] name = "icmp-socket2" version = "0.6.4" @@ -265,6 +334,31 @@ dependencies = [ "hashbrown", ] +[[package]] +name = "irtt-client" +version = "0.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a59c24704bd94f19b84b2b0190aa9163efede499f8cf4eb02922443cfee08339" +dependencies = [ + "irtt-proto", + "nix", + "socket2", + "thiserror", + "tokio", +] + +[[package]] +name = "irtt-proto" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a56ff307983b8f6477724e08cf726c5e24975b9affeb853d0830d91c60026ab2" +dependencies = [ + "hmac", + "md-5", + "subtle", + "thiserror", +] + [[package]] name = "itertools" version = "0.13.0" @@ -337,6 +431,16 @@ version = "0.4.34" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f9f8bd3e56ce4dfc153cf470fffbfa98c7620958b312ca5c3a4b8d5181fd13c6" +[[package]] +name = "md-5" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "69b6441f590336821bb897fb28fc622898ccceb1d6cea3fde5ea86b090c4de98" +dependencies = [ + "cfg-if", + "digest", +] + [[package]] name = "memchr" version = "2.8.3" @@ -679,6 +783,7 @@ dependencies = [ "fastrand", "flume", "icmp-socket2", + "irtt-client", "libc", "log", "netlink-bindings", @@ -697,6 +802,12 @@ version = "1.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "85ec932792e62b2d3936163eacbd2e09984453bcd114f665c1e1694e36ff59c9" +[[package]] +name = "subtle" +version = "2.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292" + [[package]] name = "syn" version = "2.0.119" @@ -836,6 +947,12 @@ dependencies = [ "toml", ] +[[package]] +name = "typenum" +version = "1.20.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6f5e870be6c3b371b77fe0ee0bafb859fa4964b4404c27de1d380043c4dda20" + [[package]] name = "uci-derive" version = "0.1.0" diff --git a/Cargo.toml b/Cargo.toml index 777f1c5..b4df466 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -25,6 +25,7 @@ csv = "1.4.0" fastrand = "2" flume = { version = "0.12.0", default-features = false, features = ["async", "eventual-fairness", "select"] } icmp-socket2 = { version = "0.6.4", features = ["tokio"] } +irtt-client = { version = "0.6.0", optional = true, features = ["ancillary", "tokio"] } libc = "0.2" rustix = { version = "1.1", features = ["fs", "system", "thread", "time"] } serde = { version = "1", features = ["derive"] } @@ -44,7 +45,8 @@ netlink-socket2 = { version = "0.3.5" } rust-uci = { git = "https://github.com/Lochnair/rust-uci.git", rev = "6835397f4f4ba8b69e75e727105186865c357417", optional = true, default-features = false } [features] -default = ["toml", "uci-vendored"] +default = ["irtt", "toml", "uci-vendored"] +irtt = ["dep:irtt-client"] toml = ["config/toml"] uci = ["dep:rust-uci"] uci-vendored = ["uci", "rust-uci/vendored"] diff --git a/README.md b/README.md index 725d217..1b55e7f 100644 --- a/README.md +++ b/README.md @@ -57,3 +57,23 @@ You can also build an OpenWrt package to install in your images with the SDK: # Copy the generated .ipk from bin/packages/... to your router opkg install sqm-autorate-rust_*.ipk ``` + +## Reflector files and inline UCI reflectors + +Inline `config reflector` sections supplement the default +`/etc/sqm-autorate-rust/reflectors.csv`. To use only inline reflectors, explicitly +disable file loading: + +```uci +config reflectors + option files 'none' + +config reflector + option host 'global.irtt.lochnair.net' + list protocols 'irtt' +``` + +Use `list files '/path/to/reflectors.csv'` instead of `option files 'none'` to +combine CSV files with inline reflectors. Omitting `files` keeps the default CSV; +UCI discards empty string options, so `option files ''` does not disable it. +Absent or blank `irtt_hmac` values use unauthenticated IRTT sessions. diff --git a/crates/uci-derive/src/lib.rs b/crates/uci-derive/src/lib.rs index 2329174..526f058 100644 --- a/crates/uci-derive/src/lib.rs +++ b/crates/uci-derive/src/lib.rs @@ -9,13 +9,13 @@ use syn::{Data, DeriveInput, Field, Fields, FieldsNamed, LitStr, Token, parse_ma /// Generates `UciSectionSchema` from a settings struct. /// -/// Every named field is mapped one-to-one: +/// Every named field without `#[uci(skip)]` is mapped one-to-one: /// /// ```text /// Rust field name -> UCI option name /// Rust field name -> config-rs field name /// ``` -#[proc_macro_derive(UciSection)] +#[proc_macro_derive(UciSection, attributes(uci))] pub fn derive_uci_section(input: TokenStream) -> TokenStream { let input = parse_macro_input!(input as DeriveInput); @@ -56,26 +56,28 @@ fn expand_uci_section(input: &DeriveInput) -> syn::Result>(); + }); + } let (impl_generics, type_generics, where_clause) = input.generics.split_for_impl(); @@ -96,6 +98,29 @@ fn expand_uci_section(input: &DeriveInput) -> syn::Result syn::Result { + let mut skip = false; + + for attribute in &field.attrs { + if !attribute.path().is_ident("uci") { + continue; + } + + attribute.parse_nested_meta(|meta| { + if !meta.path.is_ident("skip") { + return Err(meta.error("expected `skip`")); + } + if skip { + return Err(meta.error("duplicate `skip` attribute")); + } + skip = true; + Ok(()) + })?; + } + + Ok(skip) +} + fn expand_uci_config(input: &DeriveInput) -> syn::Result { let struct_name = &input.ident; let fields = require_named_fields(input)?; diff --git a/reflectors-icmp.csv b/reflectors-icmp.csv deleted file mode 100644 index 2e1f889..0000000 --- a/reflectors-icmp.csv +++ /dev/null @@ -1,305 +0,0 @@ -reflector_ip,ip_version,description -94.140.14.14,4,AdGuard -94.140.15.15,4,AdGuard -94.140.14.15,4,AdGuard Family -94.140.15.16,4,AdGuard Family -94.140.14.140,4,AdGuard Unfiltered -94.140.14.141,4,AdGuard Unfiltered -205.171.3.65,4,CenturyLink/Level3 -205.171.2.65,4,CenturyLink/Level3 -185.228.168.9,4,CleanBrowsing -185.228.169.9,4,CleanBrowsing -185.228.168.168,4,CleanBrowsing -185.228.169.168,4,CleanBrowsing -185.228.168.10,4,CleanBrowsing -185.228.169.11,4,CleanBrowsing -8.26.56.26,4,Comodo -8.20.247.20,4,Comodo -84.200.69.80,4,DNS.WATCH -84.200.70.40,4,DNS.WATCH -46.227.200.54,4,FAELIX Anycast -46.227.200.55,4,FAELIX Anycast -45.33.97.5,4,FreeDNS -37.235.1.177,4,FreeDNS -74.82.42.42,4,Hurricane Electric -194.242.2.2,4,Mullvad -194.242.2.3,4,Mullvad -193.19.108.2,4,Mullvad -193.19.108.3,4,Mullvad -208.67.222.222,4,OpenDNS -208.67.220.220,4,OpenDNS -208.67.222.123,4,OpenDNS Family -208.67.220.123,4,OpenDNS Family -208.67.222.2,4,OpenDNS Sbx -208.67.220.2,4,OpenDNS Sbx -159.89.120.99,4,OpenNIC -134.195.4.2,4,OpenNIC -9.9.9.9,4,Quad9 -9.9.9.10,4,Quad9 -9.9.9.11,4,Quad9 -149.112.112.10,4,Quad9 -149.112.112.11,4,Quad9 -149.112.112.112,4,Quad9 -176.9.93.198,4,dnsforge.de -176.9.1.117,4,dnsforge.de -104.236.210.29,4,DNSReactor -37.235.1.174,4,FreeDNS -81.218.119.1,4,GreenTeamDNS -194.8.57.12,4,Nuernberg Internet Exchange (N-IX) -51.89.30.51,4,PublicDNS.eu -51.89.93.25,4,PublicDNS.eu -130.59.31.248,4,SWITCH -130.59.31.251,4,SWITCH -185.49.141.37,4,GetDNS -158.64.1.29,4,Restena Foundation -89.234.186.112,4,Neutopia -185.95.218.42,4,Digitale Gesellschaft Schweiz -185.95.218.43,4,Digitale Gesellschaft Schweiz -140.238.215.192,4,post-factum -199.58.81.218,4,CMRG DNS -146.255.56.98,4,Foundation for Applied Privacy -185.222.222.222,4,dns.sb -185.184.222.222,4,dns.sb -5.1.66.255,4,Freifunk München e.V. -80.80.80.80,4,Freenom World -80.80.81.81,4,Freenom World -80.67.169.12,4,French Data Network (FDN) -80.67.169.40,4,French Data Network (FDN) -217.24.255.134,4,Albtelecom Sh.a. -213.163.127.229,4,Telekomi i Kosoves SH.A. -85.94.178.198,4,ANDORRA TELECOM SAU -186.148.128.86,4,ETERNET S.R.L. -190.151.144.21,4,Lima Video Cable S.A. (Cabletel) -200.45.184.43,4,Telecom Argentina S.A. -179.60.232.10,4,WICORP SA -170.210.83.110,4,Red de Interconexion Universitaria -190.57.234.194,4,Gigared S.A. -186.153.180.148,4,Telecom Argentina S.A. -81.16.8.110,4,Ucom CJSC -45.133.105.123,4,GNC-Alfa CJSC -61.8.0.113,4,Telstra Corporation Ltd -103.224.162.40,4,Ezi-Web -139.130.4.4,4,Telstra Corporation Ltd -139.134.5.51,4,Telstra Corporation Ltd -139.134.2.190,4,Telstra Corporation Ltd -203.50.2.71,4,Telstra Corporation Ltd -83.137.41.8,4,nemox.net Informationstechnologie OG -83.137.41.9,4,nemox.net Informationstechnologie OG -194.36.144.87,4,netcup GmbH -188.118.227.21,4,Hutchison Drei Austria GmbH -83.218.176.140,4,myNet GmbH -85.132.85.85,4,Delta Telecom Ltd -45.250.253.222,4,New Generation Internet Services Limited -103.77.188.18,4,NETSCOPE -103.77.188.19,4,NETSCOPE -103.80.1.2,4,SS Online -103.153.154.2,4,Pabna Cable Vision Dot Net -114.130.5.5,4,Tire-1 IP Transit Provider of Bangladesh -114.130.5.6,4,Tire-1 IP Transit Provider of Bangladesh -103.146.221.20,4,Kloud Technologies Limited -103.146.221.21,4,Kloud Technologies Limited -115.127.19.129,4,BRACNet Limited -115.127.19.137,4,BRACNet Limited -103.113.200.10,4,National University -177.131.114.86,4,ACESSOLINE TELECOMUNICACOES LTDA -177.66.203.10,4,Downup Telecomunicacoes e servico LTDA -177.184.131.180,4,SOBRALNET SERVICOS E TELECOMUNICACOES LTDA - ME -186.194.224.82,4,R&R PROVEDOR DE INTERNET LTDA -186.225.194.29,4,Pannet Serviços On Line Ltda -189.42.239.34,4,CLARO S.A. -200.99.138.94,4,TELEFÔNICA BRASIL S.A -200.99.138.103,4,TELEFÔNICA BRASIL S.A -200.167.191.114,4,CLARO S.A. -200.169.8.1,4,Century Telecom Ltda -141.0.102.202,4,Altibox AS -62.97.223.62,4,BKK Digitek AS -217.18.206.12,4,EVRY NORGE AS -217.18.206.22,4,EVRY NORGE AS -195.204.130.85,4,GLOBALCONNECT AS -81.175.38.1,4,GLOBALCONNECT AS -46.182.19.48,4,Martin Prager trading as NbIServ -82.96.64.2,4,Probe Networks -94.247.43.254,4,OpenNIC -195.10.195.195,4,meerfarbig GmbH & Co. KG -194.172.160.4,4,Baumann Technologie GmbH -185.233.106.232,4,netcup GmbH -185.233.107.4,4,netcup GmbH -141.1.1.1,4,Vodafone Group PLC -141.1.27.249,4,Vodafone Group PLC -176.103.130.130,4,Serveroid LLC -176.103.130.131,4,Serveroid LLC -80.243.64.67,4,PJSC Vimpelcom -195.218.133.88,4,PJSC Vimpelcom -83.69.120.1,4,PJSC Vimpelcom -46.16.229.223,4,JSC Elektrosvyaz -193.183.98.66,4,Prometeus di Daniela Agro -212.19.96.2,4,Leonet4cloud srl -217.12.181.97,4,IRIDEOS S.P.A. -62.149.128.4,4,Aruba S.p.A. -62.149.132.2,4,Aruba S.p.A. -85.47.198.82,4,Telecom Italia S.p.A. -202.136.162.11,4,ASN - NTT SINGAPORE PTE LTD -51.79.144.245,4,OVH SAS -61.46.244.211,4,Jupiter Telecommunications Co. Ltd. -1.33.184.173,4,NTT PC Communications Inc. -1.33.184.116,4,NTT PC Communications Inc. -1.33.184.174,4,NTT PC Communications Inc. -1.33.184.108,4,NTT PC Communications Inc. -1.33.184.220,4,NTT PC Communications Inc. -1.33.184.230,4,NTT PC Communications Inc. -1.33.184.158,4,NTT PC Communications Inc. -62.212.113.125,4,Nerim SAS -185.30.211.154,4,ENTREPRISE DECIMA SARL -87.240.90.128,4,KERTEL SAS -45.77.61.165,4,The Constant Company LLC -178.62.57.141,4,DigitalOcean LLC -194.145.240.6,4,M247 Ltd -195.27.1.1,4,Vodafone Group PLC -81.17.66.13,4,South West Communications Group Ltd -87.117.196.200,4,IOMART CLOUD SERVICES LIMITED -213.123.251.87,4,British Telecommunications PLC -81.149.21.254,4,British Telecommunications PLC -196.21.247.1,4,TENET (The UNINET Project) -89.140.69.49,4,VODAFONE ONO S.A. -62.97.123.107,4,COLT Technology Services Group Limited -194.177.210.210,4,National Infrastructures for Research and Technology S.A. -168.126.63.1,4,Korea Telecom -121.254.134.99,4,LG DACOM Corporation -203.239.131.1,4,ELIMNET INC. -202.57.32.49,4,A Multihomed ISP Company -81.17.31.34,4,Private Layer INC -79.141.82.250,4,NTD SA -80.254.77.39,4,Monzoon Networks AG -85.93.212.2,4,Visual Online S.A. -196.223.158.58,4,MAX NET FOR INTERNET SERVICES -89.150.195.195,4,Media Network i Halmstad AB -130.244.126.100,4,TELE2 -87.96.144.179,4,A3 Sverige AB -146.247.80.144,4,A3 Sverige AB -46.246.29.69,4,GleSYS AB -203.210.237.165,4,VNPT Corp -113.190.253.229,4,VNPT Corp -119.29.29.29,4,Shenzhen Tencent Computer Systems Company Limited -209.193.4.7,4,Alaska Communications -209.112.160.2,4,Alaska Communications -216.68.4.10,4,Cincinnati Bell ZoomTown -216.68.5.10,4,Cincinnati Bell ZoomTown -1.250.0.14,4,Verizon (Newark NJ) -1.252.0.14,4,Verizon (Reston VA) -1.252.0.12,4,Verizon (Reston VA) -8.238.120.14,4,Verizon (Atlanta GA) -8.238.120.12,4,Verizon (Atlanta GA) -78.143.192.10,4,Fast (UK) -201.220.136.186,4,CABLECOLOR S.A. -190.124.166.8,4,Redes y Telecomunicaciones -190.64.140.243,4,Administracion Nacional de Telecomunicaciones -200.40.230.36,4,Administracion Nacional de Telecomunicaciones -134.249.156.228,4,Kyivstar PJSC -195.60.70.5,4,VAK Ltd. -62.140.239.1,4,LLC TRC FIORD -94.158.153.236,4,Comfo LTD -109.254.13.190,4,Donbass Electronic Communications Ltd. -176.67.0.60,4,LLC McLaut-Invest -193.34.173.206,4,New Information Systems PP -94.153.241.134,4,Kyivstar PJSC -216.104.206.136,4,Africa Online Uganda -154.72.212.38,4,National Information Technology Authority Uganda -217.174.226.78,4,State Company of Electro Communications Turkmentelecom -46.45.187.181,4,Radore Veri Merkezi Hizmetleri A.S. -46.45.187.163,4,Radore Veri Merkezi Hizmetleri A.S. -46.45.187.184,4,Radore Veri Merkezi Hizmetleri A.S. -46.45.187.186,4,Radore Veri Merkezi Hizmetleri A.S. -46.20.159.27,4,DoraTelekom -93.113.60.84,4,Netinternet Bilisim Teknolojileri AS -93.113.60.82,4,Netinternet Bilisim Teknolojileri AS -196.3.132.153,4,Telecommunication Services of Trinidad and Tobago -196.3.132.154,4,Telecommunication Services of Trinidad and Tobago -80.248.72.1,4,CAFE Informatique et telecommunications -117.121.215.102,4,UIH -1.20.151.94,4,TOT Public Company Limited -117.121.215.101,4,UIH -122.154.230.146,4,The Communication Authoity of Thailand CAT -101.101.101.101,4,Taiwan Network Information Center -101.102.103.104,4,Taiwan Network Information Center -113.196.55.130,4,New Century InfoComm Tech Co. Ltd. -210.242.68.59,4,Data Communication Business Group -92.60.50.40,4,BONET Systems s.r.o. -217.73.17.110,4,VNET a.s. -84.245.69.193,4,SWAN a.s. -178.254.129.65,4,Drustvo za telekomunikacije Orion telekom doo Beograd-Zemun -93.86.63.54,4,TELEKOM SRBIJA a.d. -109.92.22.97,4,TELEKOM SRBIJA a.d. -197.159.166.81,4,Companhia Santomense de Telecomunicacoes -208.84.200.28,4,Columbus Communication St. Vincent and the Grenadines Ltd. -208.94.176.225,4,Karib Cable -199.116.57.181,4,IP SOLUTIONS INC. -74.120.24.129,4,IP SOLUTIONS INC. -74.120.24.81,4,IP SOLUTIONS INC. -74.120.24.97,4,IP SOLUTIONS INC. -78.130.38.222,4,NOS COMUNICACOES S.A. -195.22.25.130,4,Claranet Limited -213.241.88.98,4,Netia SA -87.204.28.12,4,Netia SA -212.160.85.232,4,Orange Polska Spolka Akcyjna -45.173.200.10,4,INVERSIONES EN TELECOMUNICACIONES DIGITALES S.A.C -200.10.231.110,4,Centro Nacional de Computacion -201.222.48.206,4,TECNOLOGIA EN ELECTRONICA E INFORMATICA SOCIEDAD ANONIMA (T.E.I.S.A) -201.222.50.30,4,TECNOLOGIA EN ELECTRONICA E INFORMATICA SOCIEDAD ANONIMA (T.E.I.S.A) -201.222.55.246,4,TECNOLOGIA EN ELECTRONICA E INFORMATICA SOCIEDAD ANONIMA (T.E.I.S.A) -103.209.52.250,4,The Professional Communications Pvt Ltd -119.63.132.26,4,Transworld Associates (Pvt.) Ltd. -83.143.8.249,4,MTN NIGERIA Communication limited -165.98.138.194,4,Amnet Telecomunicaciones S.A. -203.81.75.37,4,Myanma Posts and Telecommunications -65.18.114.254,4,Telecom International Myanmar Co. Ltd -65.18.117.174,4,Telecom International Myanmar Co. Ltd -69.160.0.86,4,OOREDOO MYANMAR -103.154.241.252,4,Myanmar Link Telecommunication Ltd -66.181.167.232,4,The first E-commerce and TriplePlay Service ISP in Mongolia. -202.21.99.42,4,Mobinet LLC. AS Mobinet Internet Service Provider -202.131.241.74,4,Mobinet LLC. AS Mobinet Internet Service Provider -124.158.93.2,4,STXCitinet. Leading Internet & VOIP Service Provider. Ulaanbaatar. Mongolia -77.89.232.86,4,ORANGE MOLDOVA S.A. -200.56.224.11,4,Marcatel Com S.A. de C.V. -201.144.135.211,4,Marcatel Com S.A. de C.V. -201.134.115.36,4,Uninet S.A. de C.V. -1.32.58.251,4,TM Net. Internet Service Provider -202.86.149.20,4,Companhia de Telecomunicacoes de Macau SARL -202.175.45.2,4,Companhia de Telecomunicacoes de Macau SARL -202.86.191.50,4,Companhia de Telecomunicacoes de Macau SARL -80.250.60.241,4,Sia Nano IT -101.78.12.50,4,Enterprise of Telecommunications Lao -168.187.161.250,4,Zajil International Telecom Company W.L.L. -62.12.114.25,4,Angani Limited -103.134.220.158,4,PT Global Media Pratama Solusindo -121.100.28.163,4,PT. Sakti Putra Mandiri -103.28.114.57,4,Lintas Data Prima PT -202.51.111.149,4,PT iForte Global Internet -103.23.20.23,4,PT Infinys System Indonesia -103.98.176.24,4,Universitas PGRI Semarang -103.98.176.143,4,Universitas PGRI Semarang -119.235.48.3,4,VAINAVI INDUSTIES LTD. INTERNET SERVICE PROVIDER. INDIA -103.194.120.33,4,GOLDY DIGINET PRIVATE LIMITED -103.194.120.3,4,GOLDY DIGINET PRIVATE LIMITED -103.194.120.129,4,GOLDY DIGINET PRIVATE LIMITED -223.31.121.171,4,Power Grid Corporation of India Limited -169.38.73.5,4,SoftLayer Technologies Inc. -202.86.251.201,4,Artech Infosystems Pvt Ltd -202.164.44.246,4,Quadrant Televentures Limited -87.229.99.1,4,SigmaNet Szolgaltato es Kereskedelm Ltd -91.137.135.75,4,OPC Networks Kft. -85.90.160.69,4,OPC Networks Kft. -195.228.219.226,4,Magyar Telekom plc. -103.6.168.114,4,NewMedia Express Pte Ltd -45.71.185.100,4,NEDETEL S.A. -186.149.243.52,4,ALTICE DOMINICANA S.A. -193.17.47.1,4,CZ.NIC z.s.p.o. -195.113.144.194,4,CZ.NIC z.s.p.o. -213.211.50.1,4,itself s.r.o. -213.211.50.2,4,itself s.r.o. -185.43.135.1,4,CZ.NIC z.s.p.o. -90.183.151.107,4,O2 Czech Republic. a.s. -27.50.20.21,4,PT. Mora Telematika Indonesia -116.193.190.164,4,PT Cloud Hosting Indonesia -103.56.148.179,4,PT Beon Intermedia -8.215.26.147,4,Alibaba.com Singapore E-Commerce Private Limited diff --git a/reflectors.csv b/reflectors.csv new file mode 100644 index 0000000..d1090cf --- /dev/null +++ b/reflectors.csv @@ -0,0 +1,307 @@ +host,country,protocols,irtt_port,irtt_hmac +94.140.14.14,,icmp-echo|icmp-timestamps,, +94.140.15.15,,icmp-echo|icmp-timestamps,, +94.140.14.15,,icmp-echo|icmp-timestamps,, +94.140.15.16,,icmp-echo|icmp-timestamps,, +94.140.14.140,,icmp-echo|icmp-timestamps,, +94.140.14.141,,icmp-echo|icmp-timestamps,, +205.171.3.65,,icmp-echo|icmp-timestamps,, +205.171.2.65,,icmp-echo|icmp-timestamps,, +185.228.168.9,,icmp-echo|icmp-timestamps,, +185.228.169.9,,icmp-echo|icmp-timestamps,, +185.228.168.168,,icmp-echo|icmp-timestamps,, +185.228.169.168,,icmp-echo|icmp-timestamps,, +185.228.168.10,,icmp-echo|icmp-timestamps,, +185.228.169.11,,icmp-echo|icmp-timestamps,, +8.26.56.26,,icmp-echo|icmp-timestamps,, +8.20.247.20,,icmp-echo|icmp-timestamps,, +84.200.69.80,DE,icmp-echo|icmp-timestamps,, +84.200.70.40,DE,icmp-echo|icmp-timestamps,, +46.227.200.54,GB,icmp-echo|icmp-timestamps,, +46.227.200.55,,icmp-echo|icmp-timestamps,, +45.33.97.5,US,icmp-echo|icmp-timestamps,, +37.235.1.177,AT,icmp-echo|icmp-timestamps,, +74.82.42.42,,icmp-echo|icmp-timestamps,, +208.67.222.222,,icmp-echo|icmp-timestamps,, +208.67.220.220,,icmp-echo|icmp-timestamps,, +208.67.222.123,,icmp-echo|icmp-timestamps,, +208.67.220.123,,icmp-echo|icmp-timestamps,, +208.67.222.2,,icmp-echo|icmp-timestamps,, +208.67.220.2,,icmp-echo|icmp-timestamps,, +159.89.120.99,CA,icmp-echo|icmp-timestamps,, +134.195.4.2,,icmp-echo|icmp-timestamps,, +9.9.9.9,,icmp-echo|icmp-timestamps,, +9.9.9.10,,icmp-echo|icmp-timestamps,, +9.9.9.11,,icmp-echo|icmp-timestamps,, +149.112.112.10,,icmp-echo|icmp-timestamps,, +149.112.112.11,,icmp-echo|icmp-timestamps,, +149.112.112.112,,icmp-echo|icmp-timestamps,, +176.9.93.198,DE,icmp-echo|icmp-timestamps,, +176.9.1.117,DE,icmp-echo|icmp-timestamps,, +104.236.210.29,US,icmp-echo|icmp-timestamps,, +37.235.1.174,AT,icmp-echo|icmp-timestamps,, +81.218.119.1,IL,icmp-echo|icmp-timestamps,, +194.8.57.12,DE,icmp-echo|icmp-timestamps,, +51.89.30.51,DE,icmp-echo|icmp-timestamps,, +51.89.93.25,DE,icmp-echo|icmp-timestamps,, +130.59.31.248,CH,icmp-echo|icmp-timestamps,, +130.59.31.251,CH,icmp-echo|icmp-timestamps,, +185.49.141.37,NL,icmp-echo|icmp-timestamps,, +158.64.1.29,LU,icmp-echo|icmp-timestamps,, +89.234.186.112,FR,icmp-echo|icmp-timestamps,, +185.95.218.42,CH,icmp-echo|icmp-timestamps,, +185.95.218.43,CH,icmp-echo|icmp-timestamps,, +140.238.215.192,CH,icmp-echo|icmp-timestamps,, +199.58.81.218,CA,icmp-echo|icmp-timestamps,, +146.255.56.98,AT,icmp-echo|icmp-timestamps,, +185.222.222.222,,icmp-echo|icmp-timestamps,, +185.184.222.222,NL,icmp-echo|icmp-timestamps,, +5.1.66.255,AT,icmp-echo|icmp-timestamps,, +80.80.80.80,,icmp-echo|icmp-timestamps,, +80.80.81.81,,icmp-echo|icmp-timestamps,, +80.67.169.12,FR,icmp-echo|icmp-timestamps,, +80.67.169.40,FR,icmp-echo|icmp-timestamps,, +217.24.255.134,AL,icmp-echo|icmp-timestamps,, +213.163.127.229,XK,icmp-echo|icmp-timestamps,, +85.94.178.198,AD,icmp-echo|icmp-timestamps,, +186.148.128.86,AR,icmp-echo|icmp-timestamps,, +190.151.144.21,AR,icmp-echo|icmp-timestamps,, +200.45.184.43,AR,icmp-echo|icmp-timestamps,, +179.60.232.10,AR,icmp-echo|icmp-timestamps,, +170.210.83.110,AR,icmp-echo|icmp-timestamps,, +190.57.234.194,AR,icmp-echo|icmp-timestamps,, +186.153.180.148,AR,icmp-echo|icmp-timestamps,, +81.16.8.110,AM,icmp-echo|icmp-timestamps,, +45.133.105.123,AM,icmp-echo|icmp-timestamps,, +61.8.0.113,AU,icmp-echo|icmp-timestamps,, +103.224.162.40,AU,icmp-echo|icmp-timestamps,, +139.130.4.4,,icmp-echo|icmp-timestamps,, +139.134.5.51,AU,icmp-echo|icmp-timestamps,, +139.134.2.190,AU,icmp-echo|icmp-timestamps,, +203.50.2.71,,icmp-echo|icmp-timestamps,, +83.137.41.8,AT,icmp-echo|icmp-timestamps,, +83.137.41.9,AT,icmp-echo|icmp-timestamps,, +194.36.144.87,DE,icmp-echo|icmp-timestamps,, +188.118.227.21,AT,icmp-echo|icmp-timestamps,, +83.218.176.140,AT,icmp-echo|icmp-timestamps,, +85.132.85.85,AZ,icmp-echo|icmp-timestamps,, +45.250.253.222,US,icmp-echo|icmp-timestamps,, +103.77.188.18,BD,icmp-echo|icmp-timestamps,, +103.77.188.19,BD,icmp-echo|icmp-timestamps,, +103.80.1.2,BD,icmp-echo|icmp-timestamps,, +103.153.154.2,BD,icmp-echo|icmp-timestamps,, +114.130.5.5,BD,icmp-echo|icmp-timestamps,, +114.130.5.6,BD,icmp-echo|icmp-timestamps,, +103.146.221.20,BD,icmp-echo|icmp-timestamps,, +103.146.221.21,BD,icmp-echo|icmp-timestamps,, +115.127.19.129,BD,icmp-echo|icmp-timestamps,, +115.127.19.137,BD,icmp-echo|icmp-timestamps,, +103.113.200.10,BD,icmp-echo|icmp-timestamps,, +177.131.114.86,BR,icmp-echo|icmp-timestamps,, +177.66.203.10,BR,icmp-echo|icmp-timestamps,, +177.184.131.180,BR,icmp-echo|icmp-timestamps,, +186.194.224.82,BR,icmp-echo|icmp-timestamps,, +186.225.194.29,BR,icmp-echo|icmp-timestamps,, +189.42.239.34,BR,icmp-echo|icmp-timestamps,, +200.99.138.94,BR,icmp-echo|icmp-timestamps,, +200.99.138.103,BR,icmp-echo|icmp-timestamps,, +200.167.191.114,BR,icmp-echo|icmp-timestamps,, +200.169.8.1,BR,icmp-echo|icmp-timestamps,, +141.0.102.202,NO,icmp-echo|icmp-timestamps,, +62.97.223.62,NO,icmp-echo|icmp-timestamps,, +217.18.206.12,NO,icmp-echo|icmp-timestamps,, +217.18.206.22,NO,icmp-echo|icmp-timestamps,, +195.204.130.85,NO,icmp-echo|icmp-timestamps,, +81.175.38.1,NO,icmp-echo|icmp-timestamps,, +46.182.19.48,DE,icmp-echo|icmp-timestamps,, +82.96.64.2,DE,icmp-echo|icmp-timestamps,, +94.247.43.254,DE,icmp-echo|icmp-timestamps,, +195.10.195.195,DE,icmp-echo|icmp-timestamps,, +194.172.160.4,DE,icmp-echo|icmp-timestamps,, +185.233.106.232,DE,icmp-echo|icmp-timestamps,, +185.233.107.4,DE,icmp-echo|icmp-timestamps,, +141.1.1.1,,icmp-echo|icmp-timestamps,, +141.1.27.249,,icmp-echo|icmp-timestamps,, +176.103.130.130,,icmp-echo|icmp-timestamps,, +176.103.130.131,,icmp-echo|icmp-timestamps,, +80.243.64.67,RU,icmp-echo|icmp-timestamps,, +195.218.133.88,RU,icmp-echo|icmp-timestamps,, +83.69.120.1,RU,icmp-echo|icmp-timestamps,, +46.16.229.223,RU,icmp-echo|icmp-timestamps,, +193.183.98.66,IT,icmp-echo|icmp-timestamps,, +212.19.96.2,IT,icmp-echo|icmp-timestamps,, +217.12.181.97,IT,icmp-echo|icmp-timestamps,, +62.149.128.4,IT,icmp-echo|icmp-timestamps,, +62.149.132.2,IT,icmp-echo|icmp-timestamps,, +85.47.198.82,IT,icmp-echo|icmp-timestamps,, +202.136.162.11,SG,icmp-echo|icmp-timestamps,, +51.79.144.245,SG,icmp-echo|icmp-timestamps,, +61.46.244.211,JP,icmp-echo|icmp-timestamps,, +1.33.184.173,JP,icmp-echo|icmp-timestamps,, +1.33.184.116,JP,icmp-echo|icmp-timestamps,, +1.33.184.174,JP,icmp-echo|icmp-timestamps,, +1.33.184.108,JP,icmp-echo|icmp-timestamps,, +1.33.184.220,JP,icmp-echo|icmp-timestamps,, +1.33.184.230,JP,icmp-echo|icmp-timestamps,, +1.33.184.158,JP,icmp-echo|icmp-timestamps,, +62.212.113.125,FR,icmp-echo|icmp-timestamps,, +185.30.211.154,FR,icmp-echo|icmp-timestamps,, +87.240.90.128,FR,icmp-echo|icmp-timestamps,, +45.77.61.165,FR,icmp-echo|icmp-timestamps,, +178.62.57.141,GB,icmp-echo|icmp-timestamps,, +194.145.240.6,GB,icmp-echo|icmp-timestamps,, +195.27.1.1,,icmp-echo|icmp-timestamps,, +81.17.66.13,GB,icmp-echo|icmp-timestamps,, +87.117.196.200,GB,icmp-echo|icmp-timestamps,, +213.123.251.87,GB,icmp-echo|icmp-timestamps,, +81.149.21.254,GB,icmp-echo|icmp-timestamps,, +196.21.247.1,ZA,icmp-echo|icmp-timestamps,, +89.140.69.49,ES,icmp-echo|icmp-timestamps,, +62.97.123.107,ES,icmp-echo|icmp-timestamps,, +194.177.210.210,GR,icmp-echo|icmp-timestamps,, +168.126.63.1,KR,icmp-echo|icmp-timestamps,, +121.254.134.99,KR,icmp-echo|icmp-timestamps,, +203.239.131.1,KR,icmp-echo|icmp-timestamps,, +202.57.32.49,PH,icmp-echo|icmp-timestamps,, +81.17.31.34,CH,icmp-echo|icmp-timestamps,, +79.141.82.250,CH,icmp-echo|icmp-timestamps,, +80.254.77.39,CH,icmp-echo|icmp-timestamps,, +85.93.212.2,LU,icmp-echo|icmp-timestamps,, +196.223.158.58,SD,icmp-echo|icmp-timestamps,, +89.150.195.195,HK,icmp-echo|icmp-timestamps,, +130.244.126.100,SE,icmp-echo|icmp-timestamps,, +87.96.144.179,SE,icmp-echo|icmp-timestamps,, +146.247.80.144,SE,icmp-echo|icmp-timestamps,, +46.246.29.69,SE,icmp-echo|icmp-timestamps,, +203.210.237.165,VN,icmp-echo|icmp-timestamps,, +113.190.253.229,VN,icmp-echo|icmp-timestamps,, +119.29.29.29,,icmp-echo|icmp-timestamps,, +209.193.4.7,US,icmp-echo|icmp-timestamps,, +209.112.160.2,US,icmp-echo|icmp-timestamps,, +216.68.4.10,US,icmp-echo|icmp-timestamps,, +216.68.5.10,US,icmp-echo|icmp-timestamps,, +1.250.0.14,KR,icmp-echo|icmp-timestamps,, +1.252.0.14,KR,icmp-echo|icmp-timestamps,, +1.252.0.12,KR,icmp-echo|icmp-timestamps,, +8.238.120.14,US,icmp-echo|icmp-timestamps,, +8.238.120.12,US,icmp-echo|icmp-timestamps,, +78.143.192.10,GB,icmp-echo|icmp-timestamps,, +201.220.136.186,HN,icmp-echo|icmp-timestamps,, +190.124.166.8,HN,icmp-echo|icmp-timestamps,, +190.64.140.243,UY,icmp-echo|icmp-timestamps,, +200.40.230.36,UY,icmp-echo|icmp-timestamps,, +134.249.156.228,UA,icmp-echo|icmp-timestamps,, +195.60.70.5,UA,icmp-echo|icmp-timestamps,, +62.140.239.1,RU,icmp-echo|icmp-timestamps,, +94.158.153.236,UA,icmp-echo|icmp-timestamps,, +109.254.13.190,UA,icmp-echo|icmp-timestamps,, +176.67.0.60,UA,icmp-echo|icmp-timestamps,, +193.34.173.206,UA,icmp-echo|icmp-timestamps,, +94.153.241.134,UA,icmp-echo|icmp-timestamps,, +216.104.206.136,UG,icmp-echo|icmp-timestamps,, +154.72.212.38,UG,icmp-echo|icmp-timestamps,, +217.174.226.78,TM,icmp-echo|icmp-timestamps,, +46.45.187.181,TR,icmp-echo|icmp-timestamps,, +46.45.187.163,TR,icmp-echo|icmp-timestamps,, +46.45.187.184,TR,icmp-echo|icmp-timestamps,, +46.45.187.186,TR,icmp-echo|icmp-timestamps,, +46.20.159.27,TR,icmp-echo|icmp-timestamps,, +93.113.60.84,TR,icmp-echo|icmp-timestamps,, +93.113.60.82,TR,icmp-echo|icmp-timestamps,, +196.3.132.153,TT,icmp-echo|icmp-timestamps,, +196.3.132.154,TT,icmp-echo|icmp-timestamps,, +80.248.72.1,TG,icmp-echo|icmp-timestamps,, +117.121.215.102,TH,icmp-echo|icmp-timestamps,, +1.20.151.94,TH,icmp-echo|icmp-timestamps,, +117.121.215.101,TH,icmp-echo|icmp-timestamps,, +122.154.230.146,TH,icmp-echo|icmp-timestamps,, +101.101.101.101,TW,icmp-echo|icmp-timestamps,, +101.102.103.104,TW,icmp-echo|icmp-timestamps,, +113.196.55.130,TW,icmp-echo|icmp-timestamps,, +210.242.68.59,TW,icmp-echo|icmp-timestamps,, +92.60.50.40,SK,icmp-echo|icmp-timestamps,, +217.73.17.110,SK,icmp-echo|icmp-timestamps,, +84.245.69.193,SK,icmp-echo|icmp-timestamps,, +178.254.129.65,RS,icmp-echo|icmp-timestamps,, +93.86.63.54,RS,icmp-echo|icmp-timestamps,, +109.92.22.97,RS,icmp-echo|icmp-timestamps,, +197.159.166.81,ST,icmp-echo|icmp-timestamps,, +208.84.200.28,VC,icmp-echo|icmp-timestamps,, +208.94.176.225,LC,icmp-echo|icmp-timestamps,, +199.116.57.181,PR,icmp-echo|icmp-timestamps,, +74.120.24.129,PR,icmp-echo|icmp-timestamps,, +74.120.24.81,PR,icmp-echo|icmp-timestamps,, +74.120.24.97,PR,icmp-echo|icmp-timestamps,, +78.130.38.222,PT,icmp-echo|icmp-timestamps,, +195.22.25.130,PT,icmp-echo|icmp-timestamps,, +213.241.88.98,PL,icmp-echo|icmp-timestamps,, +87.204.28.12,PL,icmp-echo|icmp-timestamps,, +212.160.85.232,PL,icmp-echo|icmp-timestamps,, +45.173.200.10,PE,icmp-echo|icmp-timestamps,, +200.10.231.110,PY,icmp-echo|icmp-timestamps,, +201.222.48.206,PY,icmp-echo|icmp-timestamps,, +201.222.50.30,PY,icmp-echo|icmp-timestamps,, +201.222.55.246,PY,icmp-echo|icmp-timestamps,, +103.209.52.250,PK,icmp-echo|icmp-timestamps,, +119.63.132.26,PK,icmp-echo|icmp-timestamps,, +83.143.8.249,NG,icmp-echo|icmp-timestamps,, +165.98.138.194,NI,icmp-echo|icmp-timestamps,, +203.81.75.37,MM,icmp-echo|icmp-timestamps,, +65.18.114.254,MM,icmp-echo|icmp-timestamps,, +65.18.117.174,MM,icmp-echo|icmp-timestamps,, +69.160.0.86,MM,icmp-echo|icmp-timestamps,, +103.154.241.252,MM,icmp-echo|icmp-timestamps,, +66.181.167.232,MN,icmp-echo|icmp-timestamps,, +202.21.99.42,MN,icmp-echo|icmp-timestamps,, +202.131.241.74,MN,icmp-echo|icmp-timestamps,, +124.158.93.2,SG,icmp-echo|icmp-timestamps,, +77.89.232.86,MD,icmp-echo|icmp-timestamps,, +200.56.224.11,MX,icmp-echo|icmp-timestamps,, +201.144.135.211,MX,icmp-echo|icmp-timestamps,, +201.134.115.36,MX,icmp-echo|icmp-timestamps,, +1.32.58.251,MY,icmp-echo|icmp-timestamps,, +202.86.149.20,MO,icmp-echo|icmp-timestamps,, +202.175.45.2,MO,icmp-echo|icmp-timestamps,, +202.86.191.50,MO,icmp-echo|icmp-timestamps,, +80.250.60.241,LV,icmp-echo|icmp-timestamps,, +101.78.12.50,LA,icmp-echo|icmp-timestamps,, +168.187.161.250,KW,icmp-echo|icmp-timestamps,, +62.12.114.25,KE,icmp-echo|icmp-timestamps,, +103.134.220.158,ID,icmp-echo|icmp-timestamps,, +121.100.28.163,ID,icmp-echo|icmp-timestamps,, +103.28.114.57,ID,icmp-echo|icmp-timestamps,, +202.51.111.149,ID,icmp-echo|icmp-timestamps,, +103.23.20.23,ID,icmp-echo|icmp-timestamps,, +103.98.176.24,ID,icmp-echo|icmp-timestamps,, +103.98.176.143,ID,icmp-echo|icmp-timestamps,, +119.235.48.3,IN,icmp-echo|icmp-timestamps,, +103.194.120.33,IN,icmp-echo|icmp-timestamps,, +103.194.120.3,IN,icmp-echo|icmp-timestamps,, +103.194.120.129,IN,icmp-echo|icmp-timestamps,, +223.31.121.171,IN,icmp-echo|icmp-timestamps,, +169.38.73.5,IN,icmp-echo|icmp-timestamps,, +202.86.251.201,IN,icmp-echo|icmp-timestamps,, +202.164.44.246,IN,icmp-echo|icmp-timestamps,, +87.229.99.1,HU,icmp-echo|icmp-timestamps,, +91.137.135.75,HU,icmp-echo|icmp-timestamps,, +85.90.160.69,HU,icmp-echo|icmp-timestamps,, +195.228.219.226,HU,icmp-echo|icmp-timestamps,, +103.6.168.114,SG,icmp-echo|icmp-timestamps,, +45.71.185.100,EC,icmp-echo|icmp-timestamps,, +186.149.243.52,DO,icmp-echo|icmp-timestamps,, +193.17.47.1,CZ,icmp-echo|icmp-timestamps,, +195.113.144.194,CZ,icmp-echo|icmp-timestamps,, +213.211.50.1,CZ,icmp-echo|icmp-timestamps,, +213.211.50.2,CZ,icmp-echo|icmp-timestamps,, +185.43.135.1,CZ,icmp-echo|icmp-timestamps,, +90.183.151.107,CZ,icmp-echo|icmp-timestamps,, +27.50.20.21,ID,icmp-echo|icmp-timestamps,, +116.193.190.164,ID,icmp-echo|icmp-timestamps,, +103.56.148.179,ID,icmp-echo|icmp-timestamps,, +8.215.26.147,ID,icmp-echo|icmp-timestamps,, +global.irtt.lochnair.net,,irtt,2112, +uk-cov1.irtt.lochnair.net,GB,irtt,2112, +uk-cov2.irtt.lochnair.net,GB,irtt,2112, +de-fra.irtt.lochnair.net,DE,irtt,2112, +nl-ams1.irtt.lochnair.net,NL,irtt,2112, +netperf-eu.bufferbloat.net,SE,irtt,2112, diff --git a/src/baseliner.rs b/src/baseliner.rs index 4908c43..8363050 100644 --- a/src/baseliner.rs +++ b/src/baseliner.rs @@ -6,13 +6,12 @@ // SPDX-License-Identifier: MPL-2.0 use crate::SHUTDOWN; -use crate::measurement::model::{MeasurementEvent, MeasurementObservation}; +use crate::measurement::model::{MeasurementEvent, MeasurementObservation, MeasurementStream}; use crate::metrics::{Metric, MetricsSender}; use crate::settings::Settings; use flume::{Receiver, Sender}; use log::{debug, info}; use std::collections::HashMap; -use std::net::IpAddr; use std::sync::atomic::Ordering; use std::time::Duration; use std::time::Instant; @@ -72,7 +71,7 @@ impl ReflectorState { #[derive(Clone, Debug)] pub struct ControlSnapshot { pub generated_at: Instant, - pub reflectors: HashMap, + pub reflectors: HashMap, } pub struct Baseliner { @@ -121,7 +120,7 @@ fn update_reflector_state( } fn process_reply( - reflectors: &mut HashMap, + reflectors: &mut HashMap, reply: &MeasurementObservation, start_time: Instant, slow_factor: f64, @@ -129,7 +128,7 @@ fn process_reply( ) -> (ReflectorState, bool) { let sample = EwmaStats::from_reply(reply); let state = reflectors - .entry(reply.stream.peer) + .entry(reply.stream) .or_insert_with(|| ReflectorState::new(sample, reply.observed_at)); let anomaly = update_reflector_state( state, @@ -144,7 +143,7 @@ fn process_reply( } fn control_snapshot( - reflectors: &HashMap, + reflectors: &HashMap, generated_at: Instant, ) -> ControlSnapshot { ControlSnapshot { @@ -155,7 +154,7 @@ fn control_snapshot( impl Baseliner { pub fn run(&self) -> anyhow::Result<()> { - let mut reflectors = HashMap::::new(); + let mut reflectors = HashMap::::new(); let slow_factor = ewma_factor( self.settings.advanced_settings.tick_interval, @@ -249,6 +248,7 @@ mod tests { MeasurementSource, MeasurementStream, MeasurementStreamId, OneWayClock, OneWayLatency, SignedDuration, }; + use std::net::IpAddr; use std::time::Duration; fn state( @@ -289,6 +289,7 @@ mod tests { }, started_at: received_at, observed_at: received_at, + round_trip: None, one_way: Some(OneWayLatency { uplink: SignedDuration::from_millis(up_time as i64), downlink: SignedDuration::from_millis(down_time as i64), @@ -434,7 +435,8 @@ mod tests { let received_at = start + Duration::from_secs(1); let reflector = "192.0.2.1".parse().unwrap(); let mut reflectors = HashMap::new(); - reflectors.insert(reflector, state(100.0, 100.0, 100.0, 100.0, start)); + let stream = reply(reflector, 0.0, 0.0, start).stream; + reflectors.insert(stream, state(100.0, 100.0, 100.0, 100.0, start)); let (updated, anomaly) = process_reply( &mut reflectors, @@ -450,7 +452,7 @@ mod tests { assert_close(updated.recent.down, 150.0); assert_close(updated.recent.up, 200.0); assert_eq!(updated.last_receive_at, received_at); - assert_close(reflectors[&reflector].recent.up, 200.0); + assert_close(reflectors[&stream].recent.up, 200.0); } #[test] @@ -469,14 +471,15 @@ mod tests { 0.5, ); let snapshot = control_snapshot(&reflectors, generated_at); + let stream = reply(reflector, 0.0, 0.0, received_at).stream; assert!(!anomaly); assert_eq!(snapshot.generated_at, generated_at); assert_close( - snapshot.reflectors[&reflector].baseline.down, + snapshot.reflectors[&stream].baseline.down, updated.baseline.down, ); - assert_close(snapshot.reflectors[&reflector].recent.up, updated.recent.up); - assert_eq!(snapshot.reflectors[&reflector].last_receive_at, received_at); + assert_close(snapshot.reflectors[&stream].recent.up, updated.recent.up); + assert_eq!(snapshot.reflectors[&stream].last_receive_at, received_at); } } diff --git a/src/controller.rs b/src/controller.rs index 11408a8..1573c13 100644 --- a/src/controller.rs +++ b/src/controller.rs @@ -1,100 +1,150 @@ -use std::{collections::HashMap, net::IpAddr, time::Duration}; +use std::{ + collections::HashMap, + net::SocketAddr, + sync::{Arc, RwLock}, + time::Duration, +}; use flume::{Receiver, SendError, Sender}; use thiserror::Error; -use crate::measurement::{ - icmp::IcmpBinding, - model::{MeasurementSource, MeasurementStream, MeasurementStreamId, ProbeSchedule}, +#[cfg(feature = "irtt")] +use crate::measurement::irtt::IrttBinding; +use crate::{ + measurement::{ + icmp::IcmpBinding, + model::{MeasurementSource, MeasurementStream, MeasurementStreamId, ProbeSchedule}, + }, + settings::ReflectorDefinition, }; #[derive(Debug, Error)] pub enum ControllerError { #[error("ICMP channel error")] IcmpChannelError(#[from] SendError>), + #[cfg(feature = "irtt")] + #[error("IRTT channel error")] + IrttChannelError(#[from] SendError>), } pub struct Controller { - reflector_rx: Receiver>, + reflector_rx: Receiver>, icmp_binding_tx: Sender>, - + #[cfg(feature = "irtt")] + irtt_binding_tx: Sender>, + reflectors: HashMap, + active_streams: Arc>>, + endpoint_streams: Arc>>, source: MeasurementSource, period: Duration, - next_stream_id: u64, - stream_ids: HashMap<(IpAddr, MeasurementSource), MeasurementStreamId>, + stream_ids: HashMap<(SocketAddr, MeasurementSource), MeasurementStreamId>, } impl Controller { pub fn new( - reflector_rx: Receiver>, + reflector_rx: Receiver>, icmp_binding_tx: Sender>, + #[cfg(feature = "irtt")] irtt_binding_tx: Sender>, + reflectors: HashMap, + active_streams: Arc>>, + endpoint_streams: Arc>>, + source: MeasurementSource, + period: Duration, ) -> Self { Self { reflector_rx, icmp_binding_tx, - source: MeasurementSource::IcmpTimestamp, - period: Duration::from_millis(500), + #[cfg(feature = "irtt")] + irtt_binding_tx, + reflectors, + active_streams, + endpoint_streams, + source, + period, next_stream_id: 0, stream_ids: HashMap::new(), } } - pub async fn run(mut self, initial_reflectors: Vec) -> Result<(), ControllerError> { + pub async fn run(mut self, initial_reflectors: Vec) -> Result<(), ControllerError> { self.update_reflectors(initial_reflectors).await?; - while let Ok(reflectors) = self.reflector_rx.recv_async().await { self.update_reflectors(reflectors).await?; } - Ok(()) } - fn stream_for(&mut self, peer: IpAddr) -> MeasurementStream { - let key = (peer, self.source); - + fn stream_for(&mut self, endpoint: SocketAddr) -> MeasurementStream { + let key = (endpoint, self.source); let id = *self.stream_ids.entry(key).or_insert_with(|| { let id = MeasurementStreamId(self.next_stream_id); self.next_stream_id += 1; id }); - MeasurementStream { id, - peer, + peer: endpoint.ip(), source: self.source, } } - async fn update_reflectors(&mut self, reflectors: Vec) -> Result<(), ControllerError> { - let count = reflectors.len(); - - let bindings = reflectors - .into_iter() - .enumerate() - .map(|(index, peer)| { - let stream = self.stream_for(peer); - - IcmpBinding::new( - stream, - ProbeSchedule { - period: self.period, - offset: evenly_spaced_offset(self.period, index, count), - }, + async fn update_reflectors( + &mut self, + endpoints: Vec, + ) -> Result<(), ControllerError> { + let count = endpoints.len(); + let mut icmp = Vec::new(); + #[cfg(feature = "irtt")] + let mut irtt = Vec::new(); + let mut streams = Vec::with_capacity(count); + let mut endpoint_streams = HashMap::new(); + for (index, endpoint) in endpoints.into_iter().enumerate() { + let Some(_definition) = self.reflectors.get(&endpoint) else { + continue; + }; + #[cfg(feature = "irtt")] + let (port, hmac) = { + ( + _definition.irtt_port.unwrap_or(2112), + _definition.irtt_hmac.clone(), ) - }) - .collect(); - - self.icmp_binding_tx - .send_async(bindings) - .await - .map_err(|e| ControllerError::IcmpChannelError(e)) + }; + let stream = self.stream_for(endpoint); + let schedule = ProbeSchedule { + period: self.period, + offset: evenly_spaced_offset(self.period, index, count), + }; + match self.source { + MeasurementSource::IcmpEcho | MeasurementSource::IcmpTimestamp => { + icmp.push(IcmpBinding::new(stream, schedule)) + } + #[cfg(feature = "irtt")] + MeasurementSource::Irtt => irtt.push(IrttBinding::new(stream, port, hmac)), + #[cfg(not(feature = "irtt"))] + MeasurementSource::Irtt => unreachable!("IRTT feature is checked before startup"), + _ => unreachable!("unsupported reflector measurement source"), + } + streams.push(stream); + endpoint_streams.insert(endpoint, stream); + } + *self + .active_streams + .write() + .expect("active stream lock poisoned") = streams; + *self + .endpoint_streams + .write() + .expect("endpoint stream lock poisoned") = endpoint_streams; + self.icmp_binding_tx.send_async(icmp).await?; + #[cfg(feature = "irtt")] + self.irtt_binding_tx.send_async(irtt).await?; + Ok(()) } } fn evenly_spaced_offset(period: Duration, index: usize, count: usize) -> Duration { let nanos = period.as_nanos() * index as u128 / count as u128; - Duration::new( (nanos / 1_000_000_000) as u64, (nanos % 1_000_000_000) as u32, diff --git a/src/main.rs b/src/main.rs index 560cab9..1bda3c4 100644 --- a/src/main.rs +++ b/src/main.rs @@ -22,16 +22,19 @@ mod util; use crate::baseliner::Baseliner; use crate::controller::Controller; use crate::measurement::icmp::IcmpEngine; +#[cfg(feature = "irtt")] +use crate::measurement::irtt::IrttEngine; +use crate::measurement::model::{MeasurementSource, MeasurementStream}; use crate::metrics::{Metric, Metrics, MetricsSender}; use crate::platform::{TrafficControlBackend, traffic_control_backend, warn_platform_limitations}; use crate::ratecontroller::Ratecontroller; use crate::reflector_selector::ReflectorSelector; -use crate::settings::Settings; +use crate::settings::{ReflectorDefinition, ReflectorProtocol, Settings}; use crate::util::RwLockExt; use ::log::{info, warn}; use flume::RecvTimeoutError; -use std::net::IpAddr; -use std::str::FromStr; +use std::collections::HashMap; +use std::net::{IpAddr, SocketAddr, ToSocketAddrs}; use std::sync::atomic::{AtomicBool, AtomicU32, Ordering}; use std::sync::{Arc, RwLock}; use std::thread; @@ -39,8 +42,11 @@ use std::thread::{JoinHandle, sleep}; use std::time::{Duration, Instant}; struct ReflectorSetup { - peers: Arc>>, - pool: Vec, + peers: Arc>>, + active_streams: Arc>>, + endpoint_streams: Arc>>, + pool: Vec, + reflectors: HashMap, reselection_enabled: bool, } pub static SHUTDOWN: AtomicBool = AtomicBool::new(false); @@ -134,29 +140,126 @@ fn restore_shaper( } } -fn setup_reflectors(settings: &Settings) -> anyhow::Result { - let configured = settings.load_reflectors()?; - let configured_count = configured.len(); +fn resolve_reflectors( + definitions: &[ReflectorDefinition], + protocol: ReflectorProtocol, + countries: &[String], + filter_country: bool, +) -> anyhow::Result> { + let mut resolved = HashMap::new(); + for definition in definitions { + if !definition.protocols.contains(&protocol) { + continue; + } + if protocol != ReflectorProtocol::Irtt + && definition + .host + .parse::() + .is_ok_and(|ip| ip.is_ipv6()) + { + continue; + } + if filter_country + && !countries.is_empty() + && definition.country.as_ref().is_some_and(|country| { + !countries + .iter() + .any(|allowed| allowed.eq_ignore_ascii_case(country)) + }) + { + continue; + } + let port = if protocol == ReflectorProtocol::Irtt { + definition.irtt_port.unwrap_or(2112) + } else { + 0 + }; + let endpoints = match (definition.host.as_str(), port).to_socket_addrs() { + Ok(endpoints) => endpoints, + Err(error) => { + warn!("Failed to resolve reflector {}: {error}", definition.host); + continue; + } + }; + for endpoint in endpoints { + if protocol != ReflectorProtocol::Irtt && endpoint.is_ipv6() { + continue; + } + insert_reflector(&mut resolved, endpoint, definition.clone())?; + } + } + Ok(resolved) +} - let default_reflectors = [ - IpAddr::from_str("9.9.9.9")?, - IpAddr::from_str("8.238.120.14")?, - IpAddr::from_str("74.82.42.42")?, - IpAddr::from_str("194.242.2.2")?, - IpAddr::from_str("208.67.222.222")?, - IpAddr::from_str("94.140.14.14")?, - ]; - - let reselection_enabled = configured_count > settings.advanced_settings.num_reflectors as usize; - let pool = if reselection_enabled { - configured +fn insert_reflector( + reflectors: &mut HashMap, + endpoint: SocketAddr, + definition: ReflectorDefinition, +) -> anyhow::Result<()> { + if let Some(existing) = reflectors.get_mut(&endpoint) { + let same_country = match (&existing.country, &definition.country) { + (Some(a), Some(b)) => a.eq_ignore_ascii_case(b), + _ => true, + }; + let same_protocols = existing + .protocols + .iter() + .all(|p| definition.protocols.contains(p)) + && definition + .protocols + .iter() + .all(|p| existing.protocols.contains(p)); + if !same_country + || !same_protocols + || existing.irtt_port.unwrap_or(2112) != definition.irtt_port.unwrap_or(2112) + || existing.irtt_hmac != definition.irtt_hmac + { + anyhow::bail!("conflicting reflector definitions for resolved endpoint {endpoint}"); + } + if existing.country.is_none() { + existing.country = definition.country; + } } else { - Vec::new() - }; + reflectors.insert(endpoint, definition); + } + Ok(()) +} + +fn setup_reflectors(settings: &Settings) -> anyhow::Result { + let protocol = settings.advanced_settings.measurement_type; + let configured = resolve_reflectors( + &settings.reflectors.entries, + protocol, + &settings.reflectors.countries, + true, + )?; + let bootstrap = resolve_reflectors(&settings.reflectors.bootstrap, protocol, &[], false)?; + let mut reflectors = bootstrap.clone(); + for (endpoint, definition) in &configured { + insert_reflector(&mut reflectors, *endpoint, definition.clone())?; + } + if reflectors.is_empty() { + anyhow::bail!("no usable reflectors for measurement protocol {protocol}"); + } + let mut pool = configured.keys().copied().collect::>(); + pool.sort(); + let mut initial = bootstrap.keys().copied().collect::>(); + initial.sort(); + if initial.is_empty() { + initial = pool + .iter() + .take(settings.reflectors.num_reflectors as usize) + .copied() + .collect(); + } + let reselection_enabled = pool.len() > settings.reflectors.num_reflectors as usize; Ok(ReflectorSetup { - peers: Arc::new(RwLock::new(default_reflectors.to_vec())), + peers: Arc::new(RwLock::new(initial)), + active_streams: Arc::new(RwLock::new(Vec::new())), + endpoint_streams: Arc::new(RwLock::new(HashMap::new())), pool, + reflectors, reselection_enabled, }) } @@ -219,6 +322,12 @@ fn wait_for_exit(error_rx: &flume::Receiver) -> anyhow::Result<() } fn run(settings: &Settings) -> anyhow::Result<()> { + #[cfg(not(feature = "irtt"))] + if settings.advanced_settings.measurement_type == ReflectorProtocol::Irtt { + anyhow::bail!( + "IRTT measurement selected, but this binary was built without the `irtt` feature" + ); + } let rt = tokio::runtime::Runtime::new()?; if settings.advanced_settings.dry_run { @@ -229,13 +338,18 @@ fn run(settings: &Settings) -> anyhow::Result<()> { let ReflectorSetup { peers: reflector_peers, + active_streams, + endpoint_streams, pool: reflector_pool, + reflectors, reselection_enabled, } = setup_reflectors(settings)?; let (control_snapshot_tx, control_snapshot_rx) = flume::unbounded(); let (error_tx, error_rx) = flume::unbounded::(); let (icmp_binding_tx, icmp_binding_rx) = flume::unbounded(); + #[cfg(feature = "irtt")] + let (irtt_binding_tx, irtt_binding_rx) = flume::unbounded(); let (measurement_tx, measurement_rx) = flume::unbounded(); let (reflector_tx, reflector_rx) = flume::unbounded(); let (reselect_tx, reselect_rx) = flume::bounded(1); @@ -246,18 +360,45 @@ fn run(settings: &Settings) -> anyhow::Result<()> { (None, None) }; - let controller = Controller::new(reflector_rx, icmp_binding_tx); - let engine = { + let source = match settings.advanced_settings.measurement_type { + ReflectorProtocol::IcmpEcho => MeasurementSource::IcmpEcho, + ReflectorProtocol::IcmpTimestamps => MeasurementSource::IcmpTimestamp, + ReflectorProtocol::Irtt => MeasurementSource::Irtt, + }; + let controller = Controller::new( + reflector_rx, + icmp_binding_tx, + #[cfg(feature = "irtt")] + irtt_binding_tx, + reflectors, + Arc::clone(&active_streams), + Arc::clone(&endpoint_streams), + source, + Duration::from_secs_f64(settings.advanced_settings.tick_interval), + ); + let icmp_engine = { let _guard = rt.enter(); IcmpEngine::new( start_time, &settings.network, - measurement_tx, + measurement_tx.clone(), icmp_binding_rx, )? }; + #[cfg(feature = "irtt")] + let (irtt_task, irtt_engine) = { + let _guard = rt.enter(); + IrttEngine::new( + Duration::from_secs_f64(settings.advanced_settings.tick_interval), + &settings.network, + measurement_tx, + irtt_binding_rx, + )? + }; - spawn_task(&rt, &error_tx, engine.run()); + spawn_task(&rt, &error_tx, icmp_engine.run()); + #[cfg(feature = "irtt")] + spawn_task(&rt, &error_tx, irtt_engine.run(irtt_task)); let dropped = Arc::new(AtomicU32::new(0)); @@ -329,6 +470,7 @@ fn run(settings: &Settings) -> anyhow::Result<()> { .expect("reselection snapshot receiver must exist when reselection is enabled"), reflector_peers_lock: Arc::clone(&reflector_peers), reflector_pool, + endpoint_streams, trigger_channel: reselect_rx, metrics: event_metrics, reflector_tx, @@ -342,7 +484,7 @@ fn run(settings: &Settings) -> anyhow::Result<()> { let mut ratecontroller = Ratecontroller::new( settings.clone(), control_snapshot_rx, - reflector_peers, + active_streams, reselect_tx, rate_metrics, )?; diff --git a/src/measurement/icmp.rs b/src/measurement/icmp.rs index 3d96b3b..cbdeb59 100644 --- a/src/measurement/icmp.rs +++ b/src/measurement/icmp.rs @@ -326,6 +326,7 @@ impl IcmpEngine { stream: probe.stream, started_at: probe.sent_at, observed_at, + round_trip: None, one_way: None, }) } @@ -385,6 +386,7 @@ impl IcmpEngine { stream: probe.stream, started_at: probe.sent_at, observed_at, + round_trip: None, one_way: Some(OneWayLatency { uplink: SignedDuration::from_millis(uplink_ms), downlink: SignedDuration::from_millis(downlink_ms), diff --git a/src/measurement/irtt.rs b/src/measurement/irtt.rs new file mode 100644 index 0000000..ad6b818 --- /dev/null +++ b/src/measurement/irtt.rs @@ -0,0 +1,418 @@ +use std::{collections::HashMap, net::SocketAddr}; + +use flume::{Receiver, Sender}; +use irtt_client::{ + ClientConfig, ClientEvent, NegotiationPolicy, OpenPolicy, + managed::{ + ManagedClient, ManagedClientConfig, ManagedClientHandle, ManagedClientTask, + ManagedCommandApplyError, ManagedCommandError, ManagedCompletionPolicy, ManagedConfigError, + ManagedEndReason, ManagedEvent, ManagedLifecycle, ManagedSubscribeError, + ManagedTargetConfig, ManagedTargetEndReason, ManagedTargetLifecycle, TargetAuth, TargetId, + }, +}; +use log::{debug, trace, warn}; +use thiserror::Error; +use tokio::sync::broadcast::error::RecvError; + +use crate::SHUTDOWN; +use crate::measurement::model::{ + MeasurementEvent, MeasurementLoss, MeasurementObservation, MeasurementStream, + MeasurementStreamId, OneWayClock, OneWayLatency, SignedDuration, +}; +use crate::settings::NetworkSettings; + +#[derive(Debug, Error)] +pub enum IrttError { + #[error("failed to build IRTT config")] + InvalidConfig(#[from] ManagedConfigError), + + #[error("event channel closed")] + EventChannelClosed, + + #[error("binding channel closed")] + BindingChannelClosed, + + #[error("error while reciving on channel")] + ChannelRecvError(#[from] RecvError), + + #[error("unable to subscribe to irtt events")] + SubscriptionError(#[from] ManagedSubscribeError), + + #[error("failed to submit IRTT target update")] + CommandError(#[from] ManagedCommandError), + + #[error("failed to apply IRTT target update")] + CommandApplyError(#[from] ManagedCommandApplyError), + + #[error("IRTT managed client failed: {0:?}")] + ManagedFailure(ManagedEndReason), + + #[error("IRTT managed task was abandoned")] + ManagedAbandoned, +} + +pub struct IrttBinding { + stream: MeasurementStream, + port: u16, + hmac: Option, +} + +impl IrttBinding { + pub fn new(stream: MeasurementStream, port: u16, hmac: Option) -> Self { + Self { stream, port, hmac } + } +} + +pub struct IrttEngine { + client: ManagedClientHandle, + + bindings: HashMap, + targets: Vec, + + binding_rx: Receiver>, + observation_tx: Sender, +} + +impl IrttEngine { + pub fn new( + interval: std::time::Duration, + settings: &NetworkSettings, + observation_tx: Sender, + binding_rx: Receiver>, + ) -> Result<(ManagedClientTask, Self), IrttError> { + let config = managed_client_config(interval, settings); + let (task, handle) = ManagedClient::task(config, Vec::new())?; + + Ok(( + task, + Self { + client: handle, + bindings: HashMap::new(), + targets: Vec::new(), + binding_rx, + observation_tx, + }, + )) + } + + async fn set_bindings(&mut self, bindings: Vec) -> Result<(), IrttError> { + let targets = bindings + .iter() + .map(|binding| { + let mut target = ManagedTargetConfig::new( + TargetId::new(binding.stream.id.0.to_string()), + SocketAddr::new(binding.stream.peer, binding.port).to_string(), + ); + if let Some(key) = binding.hmac.as_ref().filter(|key| !key.trim().is_empty()) { + target.auth = TargetAuth::Override(irtt_client::Authentication::Hmac( + irtt_client::HmacKey::new(key.as_bytes().to_vec().into_boxed_slice()), + )); + } + target + }) + .collect::>(); + + self.client.update_targets(targets.clone())?.await?; + self.targets = targets; + + self.bindings = bindings + .into_iter() + .map(|binding| (binding.stream.id, binding)) + .collect(); + + Ok(()) + } + + pub async fn run(mut self, task: ManagedClientTask) -> Result<(), IrttError> { + let mut task = Box::pin(task); + let client = self.client.clone(); + + // Keep driving the client even while the event loop awaits command receipts. + let result = tokio::select! { + biased; + outcome = task.as_mut() => { + return match outcome.end_reason { + ManagedEndReason::StopRequested => Ok(()), + end_reason => Err(IrttError::ManagedFailure(end_reason)), + }; + } + result = self.run_events() => result, + }; + + // Every event-loop exit stops and joins the driver, including channel errors. + let stop = client.stop(); + let (outcome, _) = tokio::join!(task, stop); + result?; + match outcome.end_reason { + ManagedEndReason::StopRequested => Ok(()), + end_reason => Err(IrttError::ManagedFailure(end_reason)), + } + } + + async fn run_events(&mut self) -> Result<(), IrttError> { + let mut events = self.client.subscribe()?; + // Durable status catches failures even if their presentation events were lost. + // Resubmitting the full desired set restarts terminal targets only. + let mut retry = tokio::time::interval(std::time::Duration::from_secs(5)); + retry.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + retry.tick().await; + loop { + tokio::select! { + bindings = self.binding_rx.recv_async() => { + let bindings = bindings + .map_err(|_| IrttError::BindingChannelClosed)?; + + self.set_bindings(bindings).await?; + } + + _ = retry.tick() => { + if SHUTDOWN.load(std::sync::atomic::Ordering::Relaxed) { + return Ok(()); + } + if self.client.status().targets.iter().any(|target| { + target.desired && target.lifecycle == ManagedTargetLifecycle::Terminal + }) { + self.client.update_targets(self.targets.clone())?.await?; + } + } + + event = events.recv() => { + let event = match event { + Ok(event) => event, + + Err(RecvError::Lagged(n)) => { + warn!("IRTT event receiver dropped {n} events"); + continue; + } + + Err(RecvError::Closed) => { + let status = self.client.status(); + + match status.lifecycle { + ManagedLifecycle::Completed => return Ok(()), + + ManagedLifecycle::Failed => { + if let Some(outcome) = &status.final_outcome { + return Err(IrttError::ManagedFailure( + outcome.end_reason.clone() + )); + } + + return Err(IrttError::ManagedAbandoned); + } + + _ => return Err(IrttError::ManagedAbandoned), + } + } + }; + + match event { + ManagedEvent::Started => { + debug!("IRTT managed client started"); + } + + ManagedEvent::TargetStateChanged { + target, + lifecycle, + } => { + debug!("IRTT target {target:?}: {lifecycle:?}"); + } + + ManagedEvent::Client { target, event } => { + let stream = target.id.as_str() + .parse::() + .ok() + .and_then(|id| { + self.bindings.get(&MeasurementStreamId(id)) + }) + .map(|binding| binding.stream); + + match event { + ClientEvent::SessionStarted(session) => { + debug!( + "IRTT session started for {}: {}", + target.id, + session.remote + ); + } + + ClientEvent::NoTestCompleted(outcome) => { + debug!( + "IRTT no-test completed for {}: {}", + target.id, + outcome.remote + ); + } + + ClientEvent::SessionClosed { .. } => { + debug!("IRTT session closed for {}", target.id); + } + + ClientEvent::EchoSent { seq, .. } => { + trace!("IRTT {} sent probe {seq}", target.id); + } + + ClientEvent::EchoReply { + sent_at, + received_at, + one_way, + rtt, + .. + } => { + let Some(stream) = stream else { + continue; + }; + + let observation = MeasurementObservation { + stream, + started_at: sent_at.mono, + observed_at: received_at.mono, + round_trip: i64::try_from(rtt.effective.as_nanos()) + .ok().map(SignedDuration::from_nanos), + one_way: one_way.and_then(map_one_way), + }; + + self.observation_tx + .send_async(MeasurementEvent::Observation(observation)) + .await + .map_err(|_| IrttError::EventChannelClosed)?; + } + + ClientEvent::EchoLoss { + sent_at, + timeout_at, + .. + } => { + let Some(stream) = stream else { + continue; + }; + + let loss = MeasurementLoss { + stream, + started_at: sent_at.mono, + detected_at: timeout_at, + }; + + self.observation_tx + .send_async(MeasurementEvent::Loss(loss)) + .await + .map_err(|_| IrttError::EventChannelClosed)?; + } + + ClientEvent::DuplicateReply { seq, .. } => { + trace!( + "IRTT {} duplicate reply {seq}", + target.id + ); + } + + ClientEvent::LateReply { seq, .. } => { + trace!( + "IRTT {} late reply {seq}", + target.id + ); + } + + ClientEvent::Warning { kind, message, .. } => { + debug!( + "IRTT {} warning ({kind:?}): {message}", + target.id + ); + } + } + } + + ManagedEvent::TargetFinished { outcome } => { + match &outcome.end_reason { + ManagedTargetEndReason::Failed(failure) => { + warn!( + "IRTT target {} failed: {failure:?}", + outcome.target.id + ); + } + + reason => { + debug!( + "IRTT target {} finished: {reason:?}", + outcome.target.id + ); + } + } + } + + ManagedEvent::Stopping => { + debug!("IRTT managed client stopping"); + } + + ManagedEvent::Completed { outcome } => { + debug!( + "IRTT managed client completed: {:?}", + outcome.end_reason + ); + return Ok(()); + } + + ManagedEvent::Failed { outcome } => { + return Err(IrttError::ManagedFailure( + outcome.end_reason.clone() + )); + } + + ManagedEvent::Abandoned => { + return Err(IrttError::ManagedAbandoned); + } + } + } + + _ = tokio::time::sleep(std::time::Duration::from_millis(100)) => { + if SHUTDOWN.load(std::sync::atomic::Ordering::Relaxed) { + return Ok(()); + } + } + } + } + } +} + +fn managed_client_config( + interval: std::time::Duration, + #[allow(unused)] settings: &NetworkSettings, +) -> ManagedClientConfig { + ManagedClientConfig { + client: ClientConfig { + socket: irtt_client::SocketConfig { + #[cfg(any(target_os = "android", target_os = "fuchsia", target_os = "linux"))] + bind_to_device: settings.measurement_bind_device.clone(), + #[cfg(any(target_os = "android", target_os = "fuchsia", target_os = "linux"))] + mark: settings.measurement_mark, + #[cfg(target_os = "freebsd")] + fib: settings.measurement_fib, + ..Default::default() + }, + request: irtt_client::SessionRequest { + duration: None, + interval, + ..Default::default() + }, + open: OpenPolicy { + negotiation: NegotiationPolicy::Strict, + ..Default::default() + }, + ..Default::default() + }, + completion: ManagedCompletionPolicy::ExplicitStop, + ..Default::default() + } +} + +fn map_one_way(sample: irtt_client::OneWayDelaySample) -> Option { + let uplink = i64::try_from(sample.client_to_server?.as_nanos()).ok()?; + + let downlink = i64::try_from(sample.server_to_client?.as_nanos()).ok()?; + + Some(OneWayLatency { + uplink: SignedDuration::from_nanos(uplink), + downlink: SignedDuration::from_nanos(downlink), + clock: OneWayClock::Uncalibrated, + }) +} diff --git a/src/measurement/mod.rs b/src/measurement/mod.rs index 0ee60f2..836dd96 100644 --- a/src/measurement/mod.rs +++ b/src/measurement/mod.rs @@ -1,2 +1,4 @@ pub(crate) mod icmp; +#[cfg(feature = "irtt")] +pub(crate) mod irtt; pub(crate) mod model; diff --git a/src/measurement/model.rs b/src/measurement/model.rs index 12b7bf5..643afec 100644 --- a/src/measurement/model.rs +++ b/src/measurement/model.rs @@ -20,7 +20,7 @@ pub enum MeasurementSource { impl Display for MeasurementSource { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { match self { - MeasurementSource::IcmpEcho => write!(f, "icmp"), + MeasurementSource::IcmpEcho => write!(f, "icmp-echo"), MeasurementSource::IcmpTimestamp => write!(f, "icmp-timestamps"), MeasurementSource::Irtt => write!(f, "irtt"), MeasurementSource::TcpSeqAck => write!(f, "tcp-seqack"), @@ -121,6 +121,9 @@ pub struct MeasurementObservation { /// Local monotonic time at which the result was observed. pub observed_at: Instant, + /// Measured round-trip latency when it differs from elapsed local time. + pub round_trip: Option, + /// Directional timing, when the measurement mechanism can provide it. pub one_way: Option, } @@ -131,7 +134,10 @@ impl MeasurementObservation { } pub fn rtt_ms(&self) -> f64 { - self.rtt().as_secs_f64() * 1000.0 + self.round_trip + .map(SignedDuration::as_secs_f64) + .unwrap_or_else(|| self.rtt().as_secs_f64()) + * 1000.0 } } diff --git a/src/ratecontroller.rs b/src/ratecontroller.rs index b099177..d35e205 100644 --- a/src/ratecontroller.rs +++ b/src/ratecontroller.rs @@ -7,6 +7,7 @@ use crate::SHUTDOWN; use crate::baseliner::ControlSnapshot; +use crate::measurement::model::MeasurementStream; use crate::metrics::{Metric, MetricsSender}; use crate::platform::{ InterfaceStatsProvider, PlatformInterfaceStats, PlatformTrafficControl, TrafficControlBackend, @@ -19,7 +20,6 @@ use flume::{Receiver, Sender}; use log::{debug, info, warn}; use std::fs::File; use std::io::Write; -use std::net::IpAddr; use std::sync::atomic::Ordering; use std::thread::sleep; use std::time::{Duration, Instant}; @@ -100,7 +100,7 @@ fn drain_latest_snapshot( fn deltas_from_snapshot( snapshot: Option<&ControlSnapshot>, - reflectors: &[IpAddr], + reflectors: &[MeasurementStream], now: Instant, tick_interval: f64, ) -> (Vec, Vec) { @@ -123,8 +123,8 @@ fn deltas_from_snapshot( up.push(up_delta); debug!( - "Reflector: {} down_delay: {} up_delay: {}", - reflector, down_delta, up_delta + "Reflector: {} ({:?}) down_delay: {} up_delay: {}", + reflector.peer, reflector.source, down_delta, up_delta ); } } @@ -139,7 +139,7 @@ pub struct Ratecontroller { settings: Settings, control_snapshot: Option, control_snapshot_rx: Receiver, - reflectors_lock: ArcRwLock>, + reflectors_lock: ArcRwLock>, reselect_trigger: Sender, state_dl: State, state_ul: State, @@ -153,7 +153,7 @@ impl Ratecontroller { fn new_with_backends( settings: Settings, control_snapshot_rx: Receiver, - reflectors_lock: ArcRwLock>, + reflectors_lock: ArcRwLock>, reselect_trigger: Sender, metrics: MetricsSender, mut stats_provider: S, @@ -491,8 +491,7 @@ impl Ratecontroller { self.state_dl.deltas = down_deltas; self.state_ul.deltas = up_deltas; - let minimum = - usize::from(self.settings.advanced_settings.num_reflectors).min(reflectors.len()); + let minimum = usize::from(self.settings.reflectors.num_reflectors).min(reflectors.len()); if reflectors.is_empty() || self.state_dl.deltas.len() < minimum || self.state_ul.deltas.len() < minimum @@ -515,7 +514,7 @@ impl Ratecontroller { pub fn new( settings: Settings, control_snapshot_rx: Receiver, - reflectors_lock: ArcRwLock>, + reflectors_lock: ArcRwLock>, reselect_trigger: Sender, metrics: MetricsSender, ) -> anyhow::Result { @@ -536,7 +535,17 @@ impl Ratecontroller { mod tests { use super::*; use crate::baseliner::{EwmaStats, ReflectorState}; + use crate::measurement::model::{MeasurementSource, MeasurementStreamId}; use std::collections::HashMap; + use std::net::IpAddr; + + fn stream(id: u64, peer: IpAddr) -> MeasurementStream { + MeasurementStream { + id: MeasurementStreamId(id), + peer, + source: MeasurementSource::IcmpTimestamp, + } + } #[test] fn fixed_seed_reproduces_initial_safe_rates() { @@ -570,7 +579,7 @@ mod tests { } fn snapshot(generated_at: Instant, marker: f64) -> ControlSnapshot { - let reflector = "192.0.2.1".parse().unwrap(); + let reflector = stream(0, "192.0.2.1".parse().unwrap()); ControlSnapshot { generated_at, reflectors: HashMap::from([( @@ -596,7 +605,9 @@ mod tests { assert!(rx.is_empty()); assert_eq!(latest.generated_at, newest_at); assert_eq!( - latest.reflectors[&"192.0.2.1".parse().unwrap()].recent.down, + latest.reflectors[&stream(0, "192.0.2.1".parse().unwrap())] + .recent + .down, 2.0 ); } @@ -604,10 +615,10 @@ mod tests { #[test] fn calculates_sorted_deltas_from_active_fresh_snapshot_state() { let now = Instant::now(); - let first: IpAddr = "192.0.2.1".parse().unwrap(); - let second: IpAddr = "192.0.2.2".parse().unwrap(); - let stale: IpAddr = "192.0.2.3".parse().unwrap(); - let inactive: IpAddr = "192.0.2.4".parse().unwrap(); + let first = stream(0, "192.0.2.1".parse().unwrap()); + let second = stream(1, "192.0.2.2".parse().unwrap()); + let stale = stream(2, "192.0.2.3".parse().unwrap()); + let inactive = stream(3, "192.0.2.4".parse().unwrap()); let snapshot = ControlSnapshot { generated_at: now, reflectors: HashMap::from([ diff --git a/src/reflector_selector.rs b/src/reflector_selector.rs index d348003..3cbbc89 100644 --- a/src/reflector_selector.rs +++ b/src/reflector_selector.rs @@ -7,12 +7,13 @@ use crate::SHUTDOWN; use crate::baseliner::ControlSnapshot; +use crate::measurement::model::MeasurementStream; use crate::metrics::{Metric, MetricsSender}; use crate::settings::Settings; use crate::util::RwLockExt; use flume::{Receiver, RecvError, RecvTimeoutError, Selector, Sender}; use log::{debug, info}; -use std::net::IpAddr; +use std::net::SocketAddr; use std::sync::atomic::Ordering; use std::sync::{Arc, RwLock}; use std::thread::sleep; @@ -98,18 +99,25 @@ fn receive_snapshots_until( } } -fn recent_rtt_for_peer(snapshot: Option<&ControlSnapshot>, peer: &IpAddr) -> Option { +fn recent_rtt_for_peer( + snapshot: Option<&ControlSnapshot>, + stream: Option<&MeasurementStream>, + now: Instant, + max_age: Duration, +) -> Option { snapshot - .and_then(|snapshot| snapshot.reflectors.get(peer)) + .and_then(|snapshot| stream.and_then(|stream| snapshot.reflectors.get(stream))) + .filter(|state| now.saturating_duration_since(state.last_receive_at) < max_age) .map(|state| (state.recent.down + state.recent.up) as u64) } pub struct ReflectorSelector { pub settings: Settings, pub snapshot_rx: Receiver, - pub reflector_peers_lock: Arc>>, - pub reflector_tx: Sender>, - pub reflector_pool: Vec, + pub reflector_peers_lock: Arc>>, + pub reflector_tx: Sender>, + pub reflector_pool: Vec, + pub endpoint_streams: Arc>>, pub trigger_channel: Receiver, pub metrics: MetricsSender, } @@ -159,13 +167,11 @@ impl ReflectorSelector { // After 40 reselections, slow down to every 15 minutes if reselection_count > 40 { - selector_sleep_time = Duration::new( - self.settings.advanced_settings.peer_reselection_time * 60, - 0, - ); + selector_sleep_time = + Duration::new(self.settings.reflectors.peer_reselection_time * 60, 0); } - let mut next_peers: Vec = Vec::new(); + let mut next_peers: Vec = Vec::new(); let mut reflectors_peers = self.reflector_peers_lock.write_anyhow()?; let current_peers = reflectors_peers.clone(); @@ -175,6 +181,9 @@ impl ReflectorSelector { next_peers.push(*reflector); } + if self.reflector_pool.is_empty() { + continue; + } for _ in 0..20 { let next_candidate = &self.reflector_pool[fastrand::usize(..self.reflector_pool.len())]; @@ -208,7 +217,13 @@ impl ReflectorSelector { let mut candidates = Vec::new(); for peer in next_peers { - if let Some(rtt) = recent_rtt_for_peer(latest_snapshot.as_ref(), &peer) { + let stream = self.endpoint_streams.read_anyhow()?.get(&peer).copied(); + if let Some(rtt) = recent_rtt_for_peer( + latest_snapshot.as_ref(), + stream.as_ref(), + Instant::now(), + Duration::from_secs_f64(self.settings.advanced_settings.tick_interval * 2.0), + ) { candidates.push((peer, rtt)); info!("Candidate reflector: {} RTT: {}", peer, rtt); } else { @@ -223,7 +238,7 @@ impl ReflectorSelector { candidates.sort_by(|a, b| a.1.cmp(&b.1)); // Now we will just limit the candidates down to 2 * num_reflectors - let mut num_reflectors = self.settings.advanced_settings.num_reflectors; + let num_reflectors = self.settings.reflectors.num_reflectors; let candidate_pool_num = (2 * num_reflectors) as usize; candidates.truncate(candidate_pool_num); @@ -245,8 +260,12 @@ impl ReflectorSelector { } if (candidates.len() as u8) < num_reflectors { - num_reflectors = candidates.len() as u8; + info!( + "Only {} usable reflectors, selecting available candidates", + candidates.len() + ); } + let num_reflectors = num_reflectors.min(candidates.len() as u8); let mut new_peers = Vec::new(); for i in 0..num_reflectors { @@ -256,7 +275,7 @@ impl ReflectorSelector { self.metrics.send(Metric::Event { name: "reflector_selected", reason: "", - reflector: Some(peer), + reflector: Some(peer.ip()), tags: &[], }); } @@ -271,7 +290,9 @@ impl ReflectorSelector { mod tests { use super::*; use crate::baseliner::{EwmaStats, ReflectorState}; + use crate::measurement::model::MeasurementSource; use std::collections::HashMap; + use std::net::IpAddr; use std::thread; fn snapshot(generated_at: Instant) -> ControlSnapshot { @@ -284,7 +305,11 @@ mod tests { fn snapshot_with_rtt(generated_at: Instant, peer: IpAddr, rtt: f64) -> ControlSnapshot { let mut reflectors = HashMap::new(); reflectors.insert( - peer, + crate::measurement::model::MeasurementStream { + id: crate::measurement::model::MeasurementStreamId(0), + peer, + source: MeasurementSource::IcmpTimestamp, + }, ReflectorState { baseline: EwmaStats { down: rtt / 2.0, @@ -384,7 +409,7 @@ mod tests { #[test] fn baseline_wait_retains_snapshot_for_evaluation() { let (snapshot_tx, snapshot_rx) = flume::unbounded(); - let peer = "192.0.2.1".parse().unwrap(); + let peer: IpAddr = "192.0.2.1".parse().unwrap(); let generated_at = Instant::now(); snapshot_tx .send(snapshot_with_rtt(generated_at, peer, 20.0)) @@ -405,7 +430,12 @@ mod tests { ); assert_eq!( - recent_rtt_for_peer(latest_snapshot.as_ref(), &peer), + recent_rtt_for_peer( + latest_snapshot.as_ref(), + latest_snapshot.as_ref().unwrap().reflectors.keys().next(), + generated_at + Duration::from_secs(1), + Duration::from_secs(2) + ), Some(42) ); } diff --git a/src/settings/mod.rs b/src/settings/mod.rs index 3539729..1aa1e5f 100644 --- a/src/settings/mod.rs +++ b/src/settings/mod.rs @@ -2,12 +2,10 @@ // // SPDX-License-Identifier: MPL-2.0 -use anyhow::{Context, bail}; use config::{Config, ConfigError, Environment}; use log::Level; use serde::{Deserialize, Deserializer, de}; use std::fmt::{self, Display}; -use std::net::IpAddr; use std::str::FromStr; use uci_derive::{UciConfig, UciSection}; @@ -48,7 +46,7 @@ where type Value = bool; fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - formatter.write_str("a boolean or a UCI-compatible boolean string") + formatter.write_str("a boolean, integer 0 or 1, or a UCI-compatible boolean string") } fn visit_bool(self, value: bool) -> Result @@ -58,6 +56,28 @@ where Ok(value) } + fn visit_i64(self, value: i64) -> Result + where + E: de::Error, + { + match value { + 0 => Ok(false), + 1 => Ok(true), + _ => Err(E::custom("boolean integer must be 0 or 1")), + } + } + + fn visit_u64(self, value: u64) -> Result + where + E: de::Error, + { + match value { + 0 => Ok(false), + 1 => Ok(true), + _ => Err(E::custom("boolean integer must be 0 or 1")), + } + } + fn visit_str(self, value: &str) -> Result where E: de::Error, @@ -79,28 +99,32 @@ where deserializer.deserialize_any(FlexiBoolVisitor) } -#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] -pub enum MeasurementType { - Icmp = 1, +#[derive(Clone, Copy, Debug, Deserialize, PartialEq, Eq, Hash)] +#[serde(rename_all = "kebab-case")] +pub enum ReflectorProtocol { + IcmpEcho = 1, IcmpTimestamps, + Irtt, } -impl Display for MeasurementType { +impl Display for ReflectorProtocol { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { match self { - MeasurementType::Icmp => write!(f, "icmp"), - MeasurementType::IcmpTimestamps => write!(f, "icmp-timestamps"), + ReflectorProtocol::IcmpEcho => write!(f, "icmp-echo"), + ReflectorProtocol::IcmpTimestamps => write!(f, "icmp-timestamps"), + ReflectorProtocol::Irtt => write!(f, "irtt"), } } } -impl FromStr for MeasurementType { +impl FromStr for ReflectorProtocol { type Err = ConfigError; fn from_str(s: &str) -> Result { match s.to_lowercase().as_str() { - "icmp" => Ok(MeasurementType::Icmp), - "icmp-timestamps" | "icmp_timestamps" => Ok(MeasurementType::IcmpTimestamps), + "icmp-echo" => Ok(ReflectorProtocol::IcmpEcho), + "icmp-timestamps" => Ok(ReflectorProtocol::IcmpTimestamps), + "irtt" => Ok(ReflectorProtocol::Irtt), _ => Err(ConfigError::Message(format!( "invalid measurement type: {s}" ))), @@ -128,12 +152,23 @@ impl FromStr for ObservabilityProtocol { } } -#[derive(Debug, Deserialize)] -struct ReflectorRecord { - reflector_ip: IpAddr, - ip_version: u8, +#[derive(Debug, Clone, Deserialize, UciSection)] +pub(crate) struct ReflectorDefinition { + pub(crate) host: String, + pub(crate) country: Option, + pub(crate) protocols: Vec, + pub(crate) irtt_port: Option, #[allow(dead_code)] - description: String, + pub(crate) irtt_hmac: Option, +} + +#[derive(Debug, Deserialize)] +struct ReflectorRow { + host: String, + country: Option, + protocols: String, + irtt_port: Option, + irtt_hmac: Option, } #[derive(Clone, Debug, Deserialize, UciConfig)] @@ -142,6 +177,10 @@ pub(crate) struct Settings { #[uci(section)] pub(crate) network: NetworkSettings, + #[serde(default)] + #[uci(section)] + pub(crate) reflectors: ReflectorSettings, + #[serde(default)] #[uci(section)] pub(crate) output: OutputSettings, @@ -160,55 +199,41 @@ impl Settings { let builder = Config::builder(); #[cfg(all(feature = "uci", unix))] - let builder = builder.add_source(UciSource::from_schema::().required(false)); + let builder = builder.add_source( + UciSource::from_schema::() + .with_repeated_schema(&[ + uci_schema::UciRepeatedSectionMapping { + uci_section: "reflector", + config_path: "reflectors.entries", + options: ::UCI_OPTIONS, + }, + uci_schema::UciRepeatedSectionMapping { + uci_section: "bootstrap_reflector", + config_path: "reflectors.bootstrap", + options: ::UCI_OPTIONS, + }, + ]) + .required(false), + ); #[cfg(feature = "toml")] let builder = builder.add_source( config::File::with_name("/etc/sqm-autorate-rust/config.toml").required(false), ); - let builder = builder.add_source(Environment::with_prefix("SQMA").separator("__")); - let built_config = builder.build()?; + let builder = builder.add_source( + Environment::with_prefix("SQMA") + .separator("__") + .try_parsing(true) + .list_separator(",") + .with_list_parse_key("reflectors.files") + .with_list_parse_key("reflectors.countries"), + ); + let mut settings: Self = builder.build()?.try_deserialize()?; - built_config.try_deserialize() - } + settings.reflectors.load_files()?; - pub fn load_reflectors(&self) -> anyhow::Result> { - let mut reader = csv::ReaderBuilder::new() - .has_headers(true) - .trim(csv::Trim::All) - .from_path(&self.advanced_settings.reflector_list_file) - .with_context(|| { - format!( - "failed to open reflector list: {}", - self.advanced_settings.reflector_list_file - ) - })?; - - let mut reflectors = Vec::with_capacity(50); - - for record in reader.deserialize::() { - let record = record.with_context(|| { - format!( - "failed to parse reflector list: {}", - self.advanced_settings.reflector_list_file - ) - })?; - - match (record.ip_version, record.reflector_ip) { - (4, IpAddr::V4(_)) => {} - (6, address @ IpAddr::V6(_)) => { - bail!("reflector {address} is IPv6; only IPv4 reflectors are supported"); - } - (version, address) => { - bail!("reflector {address} has inconsistent IP version {version}"); - } - } - - reflectors.push(record.reflector_ip); - } - - Ok(reflectors) + Ok(settings) } } @@ -241,6 +266,100 @@ impl NetworkSettings { } } +#[derive(Clone, Debug, Deserialize, UciSection)] +#[serde(default)] +pub(crate) struct ReflectorSettings { + #[serde(deserialize_with = "deserialize_reflector_files")] + pub(crate) files: Vec, + pub(crate) countries: Vec, + + #[uci(skip)] + pub(crate) entries: Vec, + + #[uci(skip)] + pub(crate) bootstrap: Vec, + + pub(crate) num_reflectors: u8, + pub(crate) peer_reselection_time: u64, +} + +impl ReflectorSettings { + fn load_files(&mut self) -> Result<(), ConfigError> { + for path in &self.files { + let mut reader = csv::Reader::from_path(path) + .map_err(|error| ConfigError::Message(error.to_string()))?; + for row in reader.deserialize::() { + let row = row.map_err(|error| ConfigError::Message(error.to_string()))?; + let protocols = row + .protocols + .split(['|', ';']) + .map(str::trim) + .filter(|protocol| !protocol.is_empty()) + .map(ReflectorProtocol::from_str) + .collect::, _>>()?; + if protocols.is_empty() { + return Err(ConfigError::Message(format!( + "reflector row in {path} has no protocols" + ))); + } + self.entries.push(ReflectorDefinition { + host: row.host, + country: row.country, + protocols, + irtt_port: row.irtt_port, + irtt_hmac: row.irtt_hmac, + }); + } + } + + Ok(()) + } +} + +impl Default for ReflectorSettings { + fn default() -> Self { + let icmp = |host: &str| ReflectorDefinition { + host: host.to_owned(), + country: None, + protocols: vec![ + ReflectorProtocol::IcmpEcho, + ReflectorProtocol::IcmpTimestamps, + ], + irtt_port: None, + irtt_hmac: None, + }; + + let irtt = |host: &str, country: Option<&str>| ReflectorDefinition { + host: host.to_owned(), + country: country.map(str::to_owned), + protocols: vec![ReflectorProtocol::Irtt], + irtt_port: Some(2112), + irtt_hmac: None, + }; + + Self { + files: vec!["/etc/sqm-autorate-rust/reflectors.csv".into()], + countries: Vec::new(), + entries: Vec::new(), + bootstrap: vec![ + icmp("9.9.9.9"), + icmp("8.238.120.14"), + icmp("74.82.42.42"), + icmp("194.242.2.2"), + icmp("208.67.222.222"), + icmp("94.140.14.14"), + irtt("global.irtt.lochnair.net", None), + irtt("uk-cov1.irtt.lochnair.net", Some("GB")), + irtt("uk-cov2.irtt.lochnair.net", Some("GB")), + irtt("de-fra.irtt.lochnair.net", Some("DE")), + irtt("nl-ams1.irtt.lochnair.net", Some("NL")), + ], + num_reflectors: 5, + peer_reselection_time: 15, + } + } +} + #[derive(Clone, Debug, Deserialize, UciSection)] #[serde(default)] pub(crate) struct OutputSettings { @@ -310,10 +429,7 @@ pub(crate) struct AdvancedSettings { pub(crate) high_load_level: f64, pub(crate) min_change_interval: f64, #[serde(deserialize_with = "deserialize_from_str")] - pub(crate) measurement_type: MeasurementType, - pub(crate) num_reflectors: u8, - pub(crate) peer_reselection_time: u64, - pub(crate) reflector_list_file: String, + pub(crate) measurement_type: ReflectorProtocol, pub(crate) speed_hist_size: u32, pub(crate) tick_interval: f64, pub(crate) upload_delay_ms: f64, @@ -327,10 +443,7 @@ impl Default for AdvancedSettings { download_delay_ms: 15.0, high_load_level: 0.8, min_change_interval: 0.5, - measurement_type: MeasurementType::IcmpTimestamps, - num_reflectors: 5, - peer_reselection_time: 15, - reflector_list_file: "/etc/sqm-autorate/reflectors-icmp.csv".to_owned(), + measurement_type: ReflectorProtocol::IcmpTimestamps, speed_hist_size: 100, tick_interval: 0.5, upload_delay_ms: 15.0, @@ -359,3 +472,24 @@ where let value = String::deserialize(deserializer)?; value.parse().map_err(de::Error::custom) } + +// UCI drops empty string options; `option files 'none'` disables file loading. +fn deserialize_reflector_files<'de, D>(deserializer: D) -> Result, D::Error> +where + D: Deserializer<'de>, +{ + #[derive(Deserialize)] + #[serde(untagged)] + enum Files { + List(Vec), + String(String), + } + + match Files::deserialize(deserializer)? { + Files::List(files) => Ok(files), + Files::String(value) if value.trim().is_empty() || value.trim() == "none" => Ok(Vec::new()), + Files::String(_) => Err(de::Error::custom( + "reflector files must be a list, an empty string, or none", + )), + } +} diff --git a/src/settings/uci.rs b/src/settings/uci.rs index 1646b67..e35d8e6 100644 --- a/src/settings/uci.rs +++ b/src/settings/uci.rs @@ -1,6 +1,6 @@ use std::path::{Path, PathBuf}; -use crate::settings::uci_schema::{UciConfigSchema, UciSectionMapping}; +use crate::settings::uci_schema::{UciConfigSchema, UciRepeatedSectionMapping, UciSectionMapping}; use config::{ConfigError as SourceError, Map, Source, Value, ValueKind}; use rust_uci::config::{ @@ -18,6 +18,7 @@ pub(crate) struct UciSource { packages: PackageSelection, required: bool, schema: &'static [UciSectionMapping], + repeated_schema: &'static [UciRepeatedSectionMapping], config_dir: Option, save_dir: Option, } @@ -28,6 +29,7 @@ impl UciSource { packages, required: true, schema: &[], + repeated_schema: &[], config_dir: None, save_dir: None, } @@ -65,6 +67,17 @@ impl UciSource { self } + /// Registers section types collected as arrays of tables at their config paths. + /// Absent section types leave the destination unset for Serde/source defaults. + #[allow(dead_code)] + pub(crate) fn with_repeated_schema( + mut self, + schema: &'static [UciRepeatedSectionMapping], + ) -> Self { + self.repeated_schema = schema; + self + } + #[allow(dead_code)] pub(crate) fn with_directories( mut self, @@ -145,6 +158,7 @@ impl UciSource { package_name: &str, package: &UciPackage, values: &mut Map, + repeated_values: &mut Map>, ) -> Result<(), SourceError> { let sections = package.sections().map_err(|e| { SourceError::Message(format!( @@ -154,6 +168,12 @@ impl UciSource { })?; for section in sections { + let repeated_mapping = self + .repeated_schema + .iter() + .find(|mapping| mapping.uci_section == section.type_()); + let mut section_values = Map::new(); + let selector = section.selector().ok_or_else(|| { SourceError::Message(format!( "loaded UCI section of type {:?} in package \ @@ -181,9 +201,21 @@ impl UciSource { let source_key = format!("{package_name}.{config_relative_key}"); - let destination_key = self - .mapped_config_key(&selector, option_name) - .unwrap_or(source_key); + let repeated_option = repeated_mapping.and_then(|mapping| { + mapping + .options + .iter() + .find(|option| option.uci_option == option_name) + }); + + let destination_key = if let Some(mapping) = repeated_option { + mapping.config_field.to_owned() + } else if repeated_mapping.is_some() { + option_name.to_owned() + } else { + self.mapped_config_key(&selector, option_name) + .unwrap_or(source_key) + }; let Some(value) = option.get().map_err(|e| { SourceError::Message(format!( @@ -210,13 +242,27 @@ impl UciSource { } }; - if values.insert(destination_key.clone(), value).is_some() { + let destination = if repeated_mapping.is_some() { + &mut section_values + } else { + &mut *values + }; + + if destination.insert(destination_key.clone(), value).is_some() { return Err(SourceError::Message(format!( "duplicate UCI destination key \ {destination_key:?} while mapping {uci_key:?}" ))); } } + + if let Some(mapping) = repeated_mapping { + let origin = format!("UCI section {package_name}.{selector}"); + repeated_values + .entry(mapping.config_path.to_owned()) + .or_default() + .push(Value::new(Some(&origin), ValueKind::Table(section_values))); + } } Ok(()) @@ -272,6 +318,7 @@ impl Source for UciSource { }; let mut values = Map::new(); + let mut repeated_values = Map::new(); for package in packages { let package_name = package @@ -279,7 +326,22 @@ impl Source for UciSource { .map_err(|e| SourceError::Message(format!("failed to read UCI package name: {e}")))? .to_owned(); - self.collect_package(&package_name, &package, &mut values)?; + self.collect_package(&package_name, &package, &mut values, &mut repeated_values)?; + } + + for (config_path, sections) in repeated_values { + let origin = "UCI repeated sections".to_owned(); + if values + .insert( + config_path.clone(), + Value::new(Some(&origin), ValueKind::Array(sections)), + ) + .is_some() + { + return Err(SourceError::Message(format!( + "duplicate UCI destination key {config_path:?} while mapping repeated sections" + ))); + } } Ok(values) diff --git a/src/settings/uci_schema.rs b/src/settings/uci_schema.rs index aedd51d..2cfd737 100644 --- a/src/settings/uci_schema.rs +++ b/src/settings/uci_schema.rs @@ -19,6 +19,18 @@ pub(crate) struct UciSectionMapping { pub(crate) options: &'static [UciOptionMapping], } +#[cfg(all(feature = "uci", unix))] +#[derive(Clone, Copy, Debug)] +pub(crate) struct UciRepeatedSectionMapping { + /// UCI section type, for both named and anonymous sections. + pub(crate) uci_section: &'static str, + + /// Config-rs / Serde path receiving an array of section tables. + pub(crate) config_path: &'static str, + + pub(crate) options: &'static [UciOptionMapping], +} + pub(crate) trait UciSectionSchema { const UCI_OPTIONS: &'static [UciOptionMapping]; } diff --git a/src/settings/validation.rs b/src/settings/validation.rs index 14a54cd..e6818a3 100644 --- a/src/settings/validation.rs +++ b/src/settings/validation.rs @@ -3,6 +3,8 @@ // SPDX-License-Identifier: MPL-2.0 // +use crate::settings::ReflectorSettings; + use super::{AdvancedSettings, NetworkSettings, ObservabilitySettings, OutputSettings, Settings}; use std::error::Error; use std::fmt::{self, Display, Formatter}; @@ -88,6 +90,7 @@ impl Settings { let mut report = ValidationReport::default(); self.network.validate(&mut report); + self.reflectors.validate(&mut report); self.output.validate(&mut report); self.observability.validate(&mut report); self.advanced_settings.validate(&mut report); @@ -192,6 +195,60 @@ impl NetworkSettings { } } +impl ReflectorSettings { + fn validate(&self, report: &mut ValidationReport) { + if self.num_reflectors == 0 { + report.error("reflectors.num_reflectors", "must be greater than 0"); + } + + if self.peer_reselection_time == 0 { + report.error("reflectors.peer_reselection_time", "must be greater than 0"); + } + + if self.files.is_empty() && self.entries.is_empty() { + report.error( + "reflectors", + "at least one reflector file or inline entry is required", + ); + } + + for (index, reflector) in self.entries.iter().enumerate() { + if reflector.host.trim().is_empty() { + report.error( + "reflectors.entries", + format!("entry {index} must have a nonempty host"), + ); + } + if reflector.protocols.is_empty() { + report.error( + "reflectors.entries", + format!("entry {index} must have at least one measurement protocol"), + ); + } + } + + for (index, file) in self.files.iter().enumerate() { + if file.trim().is_empty() { + report.error( + "reflectors.files", + format!("entry {index} must not be empty"), + ); + } + } + + for (index, country) in self.countries.iter().enumerate() { + if country.len() != 2 || !country.bytes().all(|c| c.is_ascii_alphabetic()) { + report.error( + "reflectors.countries", + format!( + "entry {index} ({country:?}) must be an ISO 3166-1 alpha-2 country code" + ), + ); + } + } + } +} + impl OutputSettings { fn validate(&self, report: &mut ValidationReport) { if self.suppress_statistics { @@ -298,24 +355,6 @@ impl AdvancedSettings { self.tick_interval, ); - if self.num_reflectors < 5 { - report.error( - "advanced_settings.num_reflectors", - format!("must be at least 5, got {}", self.num_reflectors,), - ); - } - - if self.peer_reselection_time == 0 { - report.error( - "advanced_settings.peer_reselection_time", - "must be greater than 0", - ); - } - - if self.reflector_list_file.trim().is_empty() { - report.error("advanced_settings.reflector_list_file", "must not be empty"); - } - if self.speed_hist_size == 0 { report.error( "advanced_settings.speed_hist_size",