Lochnair-Labs/sqm-autorate-rust
2
0 Fork 0
Code Issues 5 Pull Requests 2 Actions 8 Packages Projects Releases Wiki Activity

Allow selection of measurement type in config

Browse Source
This commit is contained in:
Nils Andreas Svee committed 2022-09-28 01:29:43 +02:00
1 parent d73693c25f
commit f20ff59e78
6 files changed
+105 -77
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/config.rs
+24 -2
View File
@@ -1,4 +1,4 @@
use crate::error::{ConfigParseError, MissingConfigError};
use crate::error::{ConfigParseError, InvalidMeasurementTypeError, MissingConfigError};
use log::Level;
#[cfg(feature = "uci")]
use rust_uci::Uci;
@@ -19,13 +19,29 @@ where
}
#[derive(Clone, Debug)]
pub(crate) enum AutorateType {
pub(crate) enum MeasurementType {
ICMP = 1,
ICMPTimestamps,
NTP,
TCPTimestamps,
}
impl FromStr for MeasurementType {
type Err = InvalidMeasurementTypeError;
fn from_str(s: &str) -> Result<Self, Self::Err> {
return match s.to_lowercase().as_str() {
"icmp" => Ok(MeasurementType::ICMP),
"icmp-timestamps" => Ok(MeasurementType::ICMPTimestamps),
"ntp" => Ok(MeasurementType::NTP),
"tcp-timestamps" => Ok(MeasurementType::TCPTimestamps),
&_ => Err(InvalidMeasurementTypeError {
type_: s.to_string(),
}),
};
}
}
#[derive(Clone, Debug)]
pub(crate) struct Config {
// Network section
@@ -46,6 +62,7 @@ pub(crate) struct Config {
pub(crate) download_delay_ms: f64,
pub(crate) high_load_level: f64,
pub(crate) min_change_interval: f64,
pub(crate) measurement_type: MeasurementType,
pub(crate) num_reflectors: u8,
pub(crate) reflector_list_file: String,
pub(crate) speed_hist_size: u32,
@@ -119,6 +136,11 @@ impl Config {
"sqm-autorate.@advanced_settings[0].high_load_level",
Some(0.8),
)?,
measurement_type: Self::get::<MeasurementType>(
"SQMA_MEASUREMENT_TYPE",
"sqm-autorate.@advanced_settings[0].measurement_type",
Some(MeasurementType::ICMPTimestamps),
)?,
min_change_interval: Self::get::<f64>(
"SQMA_MIN_CHANGE_INTERVAL",
"sqm-autorate.@advanced_settings[0].min_change_interval",
src/error.rs
+21 -2
View File
@@ -24,6 +24,23 @@ impl Display for ConfigParseError {
}
}
#[derive(Default, Debug)]
pub struct InvalidMeasurementTypeError {
pub(crate) type_: String,
}
impl Error for InvalidMeasurementTypeError {
fn source(&self) -> Option<&(dyn Error + 'static)> {
None
}
}
impl Display for InvalidMeasurementTypeError {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
write!(f, "Couldn't find measurement type: {}", self.type_)
}
}
#[derive(Default, Debug)]
pub struct MissingConfigError {
pub(crate) config_key: String,
@@ -63,7 +80,9 @@ impl Display for NoQdiscFoundError {
}
#[derive(Default, Debug)]
pub struct PingParseError;
pub struct PingParseError {
pub(crate) msg: String,
}
impl Error for PingParseError {
fn source(&self) -> Option<&(dyn Error + 'static)> {
@@ -74,6 +93,6 @@ impl Error for PingParseError {
// Implement std::fmt::Display for AppError
impl Display for PingParseError {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
write!(f, "Couldn't parse received ping packet") // user-facing output
write!(f, "Couldn't parse received ping packet: {}", self.msg) // user-facing output
}
}
src/main.rs
+21 -7
View File
@@ -26,11 +26,12 @@ use std::thread::sleep;
use std::time::Duration;
use std::{panic, process, thread};
use crate::config::Config;
use crate::config::{Config, MeasurementType};
use crate::log::SimpleLogger;
use crate::netlink::{Netlink, Qdisc};
use crate::pinger::{PingListener, PingSender, SocketType};
use crate::pinger_icmp::{PingerICMPEchoListener, PingerICMPEchoSender};
use crate::pinger_icmp_ts::{PingerICMPTimestampListener, PingerICMPTimestampSender};
use crate::ratecontroller::{Ratecontroller, StatsDirection};
use crate::reflector_selector::ReflectorSelector;
use crate::utils::Utils;
@@ -51,7 +52,7 @@ fn main() -> ExitCode {
}
};
crate::log::init(config.log_level).expect("Couldn't initialize logger");
log::init(config.log_level).expect("Couldn't initialize logger");
let mut reflectors = match config.load_reflectors() {
Ok(refl) => refl,
@@ -96,10 +97,22 @@ fn main() -> ExitCode {
let (baseliner_stats_sender, baseliner_stats_receiver) = channel();
let (reselect_sender, reselect_receiver) = channel();
//let mut pinger_receiver = PingerICMPTimestampListener::new(id);
//let mut pinger_sender = PingerICMPTimestampSender::new(id);
let mut pinger_receiver = PingerICMPEchoListener::new(id);
let mut pinger_sender = PingerICMPEchoSender::new(id);
let (mut pinger_receiver, mut pinger_sender) = match config.measurement_type {
MeasurementType::ICMP => (
Box::new(PingerICMPEchoListener {}) as Box<dyn PingListener + Send>,
Box::new(PingerICMPEchoSender {}) as Box<dyn PingSender + Send>,
),
MeasurementType::ICMPTimestamps => (
Box::new(PingerICMPTimestampListener {}) as Box<dyn PingListener + Send>,
Box::new(PingerICMPTimestampSender {}) as Box<dyn PingSender + Send>,
),
MeasurementType::NTP => {
todo!()
}
MeasurementType::TCPTimestamps => {
todo!()
}
};
let baseliner = Baseliner {
config: config.clone(),
@@ -137,6 +150,7 @@ fn main() -> ExitCode {
.name("receiver".to_string())
.spawn(move || {
match pinger_receiver.listen(
id,
SocketType::ICMP,
reflector_peers_lock_clone,
baseliner_stats_sender,
@@ -157,7 +171,7 @@ fn main() -> ExitCode {
let sender_handle = thread::Builder::new()
.name("sender".to_string())
.spawn(
move || match pinger_sender.send(SocketType::ICMP, reflector_peers_lock_clone) {
move || match pinger_sender.send(id, SocketType::ICMP, reflector_peers_lock_clone) {
Ok(_) => {}
Err(e) => {
error!("Error occured in sender thread: {}", e);
src/pinger.rs
+13 -12
View File
@@ -8,7 +8,7 @@ use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddrV4, SocketAddrV6};
use std::os::unix::io::RawFd;
use std::str::FromStr;
use std::sync::mpsc::Sender;
use std::sync::{Arc, Mutex, RwLock};
use std::sync::{Arc, RwLock};
use std::thread;
use std::time::Duration;
@@ -53,11 +53,9 @@ fn open_socket(type_: SocketType) -> nix::Result<RawFd> {
}
pub trait PingListener {
fn new(id: u16) -> Self;
fn get_id(&self) -> u16;
fn listen(
&mut self,
id: u16,
type_: SocketType,
reflectors_lock: Arc<RwLock<Vec<IpAddr>>>,
stats_sender: Sender<PingReply>,
@@ -85,7 +83,7 @@ pub trait PingListener {
continue;
}
let reply = match Self::parse_packet(addr, buf, size) {
let reply = match self.parse_packet(id, addr, buf, size) {
Ok(val) => val,
Err(e) => {
error!(
@@ -101,16 +99,19 @@ pub trait PingListener {
}
}
fn parse_packet(reflector: IpAddr, buf: &[u8], len: usize)
-> Result<PingReply, Box<dyn Error>>;
fn parse_packet(
&self,
id: u16,
reflector: IpAddr,
buf: &[u8],
len: usize,
) -> Result<PingReply, Box<dyn Error>>;
}
pub trait PingSender {
fn new(id: u16) -> Self;
fn get_id(&self) -> u16;
fn send(
&mut self,
id: u16,
type_: SocketType,
reflectors_lock: Arc<RwLock<Vec<IpAddr>>>,
) -> Result<(), Box<dyn Error>> {
@@ -141,7 +142,7 @@ pub trait PingSender {
}
}
let buf_v = self.craft_packet(seq);
let buf_v = self.craft_packet(id, seq);
let buf = buf_v.as_slice();
sendto(socket, buf, addr.as_ref(), MsgFlags::empty()).expect("Couldn't send ping");
@@ -157,5 +158,5 @@ pub trait PingSender {
}
}
fn craft_packet(&self, seq: u16) -> Vec<u8>;
fn craft_packet(&self, id: u16, seq: u16) -> Vec<u8>;
}
src/pinger_icmp.rs
+9 -28
View File
@@ -9,25 +9,15 @@ use etherparse::{IcmpEchoHeader, Icmpv4Header, Icmpv4Type, SlicedPacket};
use nix::sys::time::TimeValLike;
use nix::time::{clock_gettime, ClockId};
pub struct PingerICMPEchoListener {
id: u16,
}
pub struct PingerICMPEchoListener {}
pub struct PingerICMPEchoSender {
id: u16,
}
pub struct PingerICMPEchoSender {}
impl PingListener for PingerICMPEchoListener {
fn new(id: u16) -> Self {
PingerICMPEchoListener { id }
}
fn get_id(&self) -> u16 {
self.id
}
// Result: RTT, down time, up time
fn parse_packet(
&self,
id: u16,
reflector: IpAddr,
buf: &[u8],
len: usize,
@@ -70,30 +60,21 @@ impl PingListener for PingerICMPEchoListener {
},
}
Err(Box::new(PingParseError {}))
Err(Box::new(PingParseError {
msg: "Reached end of parsing function".to_string(),
}))
}
}
impl PingSender for PingerICMPEchoSender {
fn new(id: u16) -> Self {
PingerICMPEchoSender { id }
}
fn get_id(&self) -> u16 {
self.id
}
fn craft_packet(&self, seq: u16) -> Vec<u8> {
fn craft_packet(&self, id: u16, seq: u16) -> Vec<u8> {
let time = clock_gettime(ClockId::CLOCK_MONOTONIC).unwrap();
let time_u64: u64 = time.num_milliseconds() as u64;
let payload = time_u64.to_ne_bytes();
// Construct a header with checksum based on the payload
let hdr = Icmpv4Header::with_checksum(
Icmpv4Type::EchoRequest(IcmpEchoHeader {
id: self.get_id(),
seq,
}),
Icmpv4Type::EchoRequest(IcmpEchoHeader { id, seq }),
&payload,
);
src/pinger_icmp_ts.rs
+17 -26
View File
@@ -7,36 +7,33 @@ use crate::Utils;
use etherparse::icmpv4::TimestampMessage;
use etherparse::TransportSlice::Icmpv4;
use etherparse::{Icmpv4Header, Icmpv4Type, SlicedPacket};
use log::warn;
use nix::time::{clock_gettime, ClockId};
pub struct PingerICMPTimestampListener {
id: u16,
}
pub struct PingerICMPTimestampListener {}
pub struct PingerICMPTimestampSender {
id: u16,
}
pub struct PingerICMPTimestampSender {}
impl PingListener for PingerICMPTimestampListener {
fn new(id: u16) -> Self {
PingerICMPTimestampListener { id }
}
fn get_id(&self) -> u16 {
self.id
}
// Result: RTT, down time, up time
fn parse_packet(
&self,
id: u16,
reflector: IpAddr,
buf: &[u8],
len: usize,
) -> Result<PingReply, Box<dyn Error>> {
match SlicedPacket::from_ip(buf) {
Err(value) => println!("Err {:?}", value),
Err(value) => warn!("Error parsing packet: {:?}", value),
Ok(value) => match value.transport {
Some(Icmpv4(icmp)) => match icmp.icmp_type() {
Icmpv4Type::TimestampReply(reply) => {
if reply.id != id {
return Err(Box::new(PingParseError {
msg: "Wrong ID".to_string(),
}));
}
let time_now = clock_gettime(ClockId::CLOCK_REALTIME).unwrap();
let time_since_midnight: i64 = (time_now.tv_sec() as i64 % 86400 * 1000)
+ (time_now.tv_nsec() as i64 / 1000000);
@@ -70,20 +67,14 @@ impl PingListener for PingerICMPTimestampListener {
},
}
Err(Box::new(PingParseError {}))
Err(Box::new(PingParseError {
msg: "Reached end of parsing function".to_string(),
}))
}
}
impl PingSender for PingerICMPTimestampSender {
fn new(id: u16) -> Self {
PingerICMPTimestampSender { id }
}
fn get_id(&self) -> u16 {
self.id
}
fn craft_packet(&self, seq: u16) -> Vec<u8> {
fn craft_packet(&self, id: u16, seq: u16) -> Vec<u8> {
let time = clock_gettime(ClockId::CLOCK_REALTIME).unwrap();
let time_since_midnight: u32 =
((time.tv_sec() % 86400 * 1000) + (time.tv_nsec() / 1000000)) as u32;
@@ -93,7 +84,7 @@ impl PingSender for PingerICMPTimestampSender {
// Construct a header with checksum based on the payload
let hdr = Icmpv4Header::with_checksum(
Icmpv4Type::TimestampRequest(TimestampMessage {
id: self.id,
id,
seq,
originate_timestamp: time_since_midnight,
receive_timestamp: 0,