pub mod datastore;
pub mod sync;
pub mod verify;
+pub mod traffic_control;
const SUBDIRS: SubdirMap = &[
("datastore", &datastore::ROUTER),
("sync", &sync::ROUTER),
+ ("traffic-control", &traffic_control::ROUTER),
("verify", &verify::ROUTER)
];
initialize_rrd_cache, rrd_update_gauge, rrd_update_derive, rrd_sync_journal,
};
use proxmox_backup::{
+ TRAFFIC_CONTROL_CACHE,
server::{
auth::check_pbs_auth,
jobstate::{
use proxmox_backup::api2::tape::backup::do_tape_backup_job;
use proxmox_backup::server::do_verification_job;
use proxmox_backup::server::do_prune_job;
-use proxmox_backup::TrafficControlCache;
fn main() -> Result<(), Error> {
proxmox_backup::tools::setup_safe_path_env();
start_task_scheduler();
start_stat_generator();
+ start_traffic_control_updater();
server.await?;
log::info!("server shutting down, waiting for active workers to complete");
tokio::spawn(task.map(|_| ()));
}
+fn start_traffic_control_updater() {
+ let abort_future = proxmox_rest_server::shutdown_future();
+ let future = Box::pin(run_traffic_control_updater());
+ let task = futures::future::select(future, abort_future);
+ tokio::spawn(task.map(|_| ()));
+}
+
use std::time::{SystemTime, Instant, Duration, UNIX_EPOCH};
fn next_minute() -> Result<Instant, Error> {
// Test WITH
// proxmox-backup-client restore vm/201/2021-10-22T09:55:56Z drive-scsi0.img img1.img --repository localhost:store2
-lazy_static::lazy_static!{
- static ref TRAFFIC_CONTROL_CACHE: Arc<Mutex<TrafficControlCache>> =
- Arc::new(Mutex::new(TrafficControlCache::new()));
+async fn run_traffic_control_updater() {
+
+ loop {
+ let delay_target = Instant::now() + Duration::from_secs(1);
+
+ {
+ let mut cache = TRAFFIC_CONTROL_CACHE.lock().unwrap();
+ cache.compute_current_rates();
+ }
+
+ tokio::time::sleep_until(tokio::time::Instant::from_std(delay_target)).await;
+ }
+
}
fn lookup_rate_limiter(
use pbs_api_types::TRAFFIC_CONTROL_ID_SCHEMA;
use proxmox_backup::api2;
+use proxmox_backup::client_helpers::connect_to_localhost;
#[api(
Ok(Value::Null)
}
+#[api(
+ input: {
+ properties: {
+ "output-format": {
+ schema: OUTPUT_FORMAT,
+ optional: true,
+ },
+ }
+ }
+)]
+/// Show current traffic for all rules.
+async fn show_current_traffic(param: Value) -> Result<Value, Error> {
+
+ let output_format = get_output_format(¶m);
+
+ let client = connect_to_localhost()?;
+
+ let mut result = client.get(&"api2/json/admin/traffic-control", Some(param)).await?;
+
+ let mut data = result["data"].take();
+
+ let info = &api2::admin::traffic_control::API_METHOD_SHOW_CURRENT_TRAFFIC;
+
+ let options = default_table_format_options()
+ .column(ColumnConfig::new("name"))
+ .column(ColumnConfig::new("rate-in"))
+ .column(ColumnConfig::new("rate-out"));
+
+ format_and_print_result_full(&mut data, &info.returns, &output_format, &options);
+
+ Ok(Value::Null)
+}
+
pub fn traffic_control_commands() -> CommandLineInterface {
let cmd_def = CliCommandMap::new()
.insert("list", CliCommand::new(&API_METHOD_LIST_TRAFFIC_CONTROLS))
+ .insert("traffic", CliCommand::new(&API_METHOD_SHOW_CURRENT_TRAFFIC))
.insert(
"show",
CliCommand::new(&API_METHOD_SHOW_TRAFFIC_CONTROL)
use std::sync::{Arc, Mutex};
use std::collections::HashMap;
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
+use std::time::Instant;
+use std::convert::TryInto;
use anyhow::Error;
use cidr::IpInet;
use super::SharedRateLimiter;
+lazy_static::lazy_static!{
+ pub static ref TRAFFIC_CONTROL_CACHE: Arc<Mutex<TrafficControlCache>> =
+ Arc::new(Mutex::new(TrafficControlCache::new()));
+}
+
struct ParsedTcRule {
config: TrafficControlRule, // original rule config
networks: Vec<IpInet>, // parsed networks
timeframe: Vec<DailyDuration>, // parsed timeframe
}
+pub struct TrafficStat {
+ pub traffic_in: u64,
+ pub rate_in: u64,
+ pub traffic_out: u64,
+ pub rate_out: u64,
+}
+
pub struct TrafficControlCache {
use_shared_memory: bool,
+ last_rate_compute: Instant,
+ current_rate_map: HashMap<String, TrafficStat>,
last_update: i64,
last_traffic_control_generation: usize,
rules: Vec<ParsedTcRule>,
last_traffic_control_generation: 0,
last_update: 0,
use_utc: false,
+ last_rate_compute: Instant::now(),
+ current_rate_map: HashMap::new(),
}
}
self.update_config(&config)
}
+ pub fn compute_current_rates(&mut self) {
+
+ let elapsed = self.last_rate_compute.elapsed().as_micros();
+ if elapsed < 200_000 { return } // not enough data
+
+ let mut new_rate_map = HashMap::new();
+
+ for (rule, (read_limit, write_limit)) in self.limiter_map.iter() {
+ let traffic_in = read_limit.as_ref().map(|l| l.traffic()).unwrap_or(0);
+ let traffic_out = write_limit.as_ref().map(|l| l.traffic()).unwrap_or(0);
+
+ let traffic_diff_in;
+ let traffic_diff_out;
+
+ if let Some(stat) = self.current_rate_map.get(rule) {
+ traffic_diff_in = traffic_in.saturating_sub(stat.traffic_in);
+ traffic_diff_out = traffic_out.saturating_sub(stat.traffic_out);
+ } else {
+ traffic_diff_in = 0;
+ traffic_diff_out = 0;
+ }
+
+ let rate_in = ((traffic_diff_in as u128) * 1_000_000) / elapsed;
+ let rate_out = ((traffic_diff_out as u128) * 1_000_000) / elapsed;
+
+ let stat = TrafficStat {
+ traffic_in,
+ traffic_out,
+ rate_in: rate_in.try_into().unwrap_or(u64::MAX),
+ rate_out: rate_out.try_into().unwrap_or(u64::MAX),
+ };
+ new_rate_map.insert(rule.clone(), stat);
+ }
+
+ self.current_rate_map = new_rate_map;
+
+ self.last_rate_compute = Instant::now()
+ }
+
+ pub fn current_rate_map(&self) -> &HashMap<String, TrafficStat> {
+ &self.current_rate_map
+ }
fn update_config(&mut self, config: &SectionConfigData) -> Result<(), Error> {
self.limiter_map.retain(|key, _value| config.sections.contains_key(key));
pub use shared_rate_limiter::SharedRateLimiter;
mod cached_traffic_control;
-pub use cached_traffic_control::TrafficControlCache;
+pub use cached_traffic_control::{TrafficControlCache, TRAFFIC_CONTROL_CACHE};
/// Get the server's certificate info (from `proxy.pem`).