use serde_json::{json, Value};
use http::{Method, HeaderMap};
-use proxmox::sys::linux::socket::set_tcp_keepalive;
-use proxmox::tools::fs::CreateOptions;
+use proxmox_sys::linux::socket::set_tcp_keepalive;
+use proxmox_sys::fs::CreateOptions;
use proxmox_lang::try_block;
use proxmox_router::{RpcEnvironment, RpcEnvironmentType, UserInformation};
+use proxmox_http::client::{RateLimitedStream, ShareableRateLimit};
+use proxmox_sys::{task_log, task_warn};
+use proxmox_sys::logrotate::LogRotate;
-use pbs_tools::{task_log, task_warn};
use pbs_datastore::DataStore;
-use proxmox_rrd::DST;
use proxmox_rest_server::{
rotate_task_log_archive, extract_cookie , AuthError, ApiConfig, RestServer, RestEnvironment,
ServerAdapter, WorkerTask, cleanup_old_tasks,
};
+use proxmox_backup::rrd_cache::{
+ initialize_rrd_cache, rrd_update_gauge, rrd_update_derive, rrd_sync_journal,
+};
use proxmox_backup::{
- get_rrd_cache, initialize_rrd_cache,
+ TRAFFIC_CONTROL_CACHE,
server::{
auth::check_pbs_auth,
jobstate::{
};
use pbs_buildcfg::configdir;
-use proxmox_systemd::time::{compute_next_event, parse_calendar_event};
-use pbs_tools::logrotate::LogRotate;
+use proxmox_time::CalendarEvent;
use pbs_api_types::{
Authid, TapeBackupJobConfig, VerificationJobConfig, SyncJobConfig, DataStoreConfig,
PROXMOX_BACKUP_TCP_KEEPALIVE_TIME,
disks::{
DiskManage,
- zfs_pool_stats,
- get_pool_from_dataset,
+ zfs_dataset_stats,
},
};
use proxmox_backup::server::do_prune_job;
fn main() -> Result<(), Error> {
+ pbs_tools::setup_libc_malloc_opts();
+
proxmox_backup::tools::setup_safe_path_env();
let backup_uid = pbs_config::backup_user()?.uid;
bail!("proxy not running as backup user or group (got uid {} gid {})", running_uid, running_gid);
}
- pbs_runtime::main(run())
+ proxmox_async::runtime::main(run())
}
None => (None, None),
};
- let nodename = proxmox::tools::nodename();
+ let nodename = proxmox_sys::nodename();
let user = userid.as_ref().map(|u| u.as_str()).unwrap_or("");
let csrf_token = csrf_token.unwrap_or_else(|| String::from(""));
start_task_scheduler();
start_stat_generator();
+ start_traffic_control_updater();
server.await?;
log::info!("server shutting down, waiting for active workers to complete");
let key_path = configdir!("/proxy.key");
let cert_path = configdir!("/proxy.pem");
+ let (config, _) = proxmox_backup::config::node::config()?;
+ let ciphers_tls_1_3 = config.ciphers_tls_1_3;
+ let ciphers_tls_1_2 = config.ciphers_tls_1_2;
+
let mut acceptor = SslAcceptor::mozilla_intermediate_v5(SslMethod::tls()).unwrap();
+ if let Some(ciphers) = ciphers_tls_1_3.as_deref() {
+ acceptor.set_ciphersuites(ciphers)?;
+ }
+ if let Some(ciphers) = ciphers_tls_1_2.as_deref() {
+ acceptor.set_cipher_list(ciphers)?;
+ }
acceptor.set_private_key_file(key_path, SslFiletype::PEM)
.map_err(|err| format_err!("unable to read proxy key {} - {}", key_path, err))?;
acceptor.set_certificate_chain_file(cert_path)
.map_err(|err| format_err!("unable to read proxy cert {} - {}", cert_path, err))?;
+ acceptor.set_options(openssl::ssl::SslOptions::NO_RENEGOTIATION);
acceptor.check_private_key().unwrap();
Ok(acceptor.build())
}
type ClientStreamResult =
- Result<std::pin::Pin<Box<tokio_openssl::SslStream<tokio::net::TcpStream>>>, Error>;
+ Result<std::pin::Pin<Box<tokio_openssl::SslStream<RateLimitedStream<tokio::net::TcpStream>>>>, Error>;
const MAX_PENDING_ACCEPTS: usize = 1024;
fn accept_connections(
sock.set_nodelay(true).unwrap();
let _ = set_tcp_keepalive(sock.as_raw_fd(), PROXMOX_BACKUP_TCP_KEEPALIVE_TIME);
+ let peer = sock.peer_addr().ok();
+ let sock = RateLimitedStream::with_limiter_update_cb(sock, move || lookup_rate_limiter(peer));
+
let ssl = { // limit acceptor_guard scope
// Acceptor can be reloaded using the command socket "reload-certificate" command
let acceptor_guard = acceptor.lock().unwrap();
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> {
None => continue,
};
- let event = match parse_calendar_event(&event_str) {
+ let event: CalendarEvent = match event_str.parse() {
Ok(event) => event,
Err(err) => {
eprintln!("unable to parse schedule '{}' - {}", event_str, err);
}
};
- let next = match compute_next_event(&event, last, false) {
+ let next = match event.compute_next_event(last) {
Ok(Some(next)) => next,
Ok(None) => continue,
Err(err) => {
let worker_type = "verificationjob";
let auth_id = Authid::root_auth_id().clone();
if check_schedule(worker_type, &event_str, &job_id) {
- let job = match Job::new(&worker_type, &job_id) {
+ let job = match Job::new(worker_type, &job_id) {
Ok(job) => job,
Err(_) => continue, // could not get lock
};
let worker_type = "tape-backup-job";
let auth_id = Authid::root_auth_id().clone();
if check_schedule(worker_type, &event_str, &job_id) {
- let job = match Job::new(&worker_type, &job_id) {
+ let job = match Job::new(worker_type, &job_id) {
Ok(job) => job,
Err(_) => continue, // could not get lock
};
let result = try_block!({
let max_size = 512 * 1024 - 1; // an entry has ~ 100b, so > 5000 entries/file
let max_files = 20; // times twenty files gives > 100000 task entries
- let has_rotated = rotate_task_log_archive(max_size, true, Some(max_files))?;
+
+ let user = pbs_config::backup_user()?;
+ let options = proxmox_sys::fs::CreateOptions::new()
+ .owner(user.uid)
+ .group(user.gid);
+
+ let has_rotated = rotate_task_log_archive(
+ max_size,
+ true,
+ Some(max_files),
+ Some(options.clone()),
+ )?;
+
if has_rotated {
task_log!(worker, "task log archive was rotated");
} else {
let max_size = 32 * 1024 * 1024 - 1;
let max_files = 14;
- let mut logrotate = LogRotate::new(pbs_buildcfg::API_ACCESS_LOG_FN, true)
- .ok_or_else(|| format_err!("could not get API access log file names"))?;
- if logrotate.rotate(max_size, None, Some(max_files))? {
+
+ let mut logrotate = LogRotate::new(
+ pbs_buildcfg::API_ACCESS_LOG_FN,
+ true,
+ Some(max_files),
+ Some(options.clone()),
+ )?;
+
+ if logrotate.rotate(max_size)? {
println!("rotated access log, telling daemons to re-open log file");
- pbs_runtime::block_on(command_reopen_access_logfiles())?;
+ proxmox_async::runtime::block_on(command_reopen_access_logfiles())?;
task_log!(worker, "API access log was rotated");
} else {
task_log!(worker, "API access log was not rotated");
}
- let mut logrotate = LogRotate::new(pbs_buildcfg::API_AUTH_LOG_FN, true)
- .ok_or_else(|| format_err!("could not get API auth log file names"))?;
+ let mut logrotate = LogRotate::new(
+ pbs_buildcfg::API_AUTH_LOG_FN,
+ true,
+ Some(max_files),
+ Some(options),
+ )?;
- if logrotate.rotate(max_size, None, Some(max_files))? {
+ if logrotate.rotate(max_size)? {
println!("rotated auth log, telling daemons to re-open log file");
- pbs_runtime::block_on(command_reopen_auth_logfiles())?;
+ proxmox_async::runtime::block_on(command_reopen_auth_logfiles())?;
task_log!(worker, "API authentication log was rotated");
} else {
task_log!(worker, "API authentication log was not rotated");
generate_host_stats().await;
+ rrd_sync_journal();
+
tokio::time::sleep_until(tokio::time::Instant::from_std(delay_target)).await;
}
}
-fn rrd_update_gauge(name: &str, value: f64) {
- if let Ok(rrd_cache) = get_rrd_cache() {
- if let Err(err) = rrd_cache.update_value(name, value, DST::Gauge) {
- eprintln!("rrd::update_value '{}' failed - {}", name, err);
- }
- }
-}
-
-fn rrd_update_derive(name: &str, value: f64) {
- if let Ok(rrd_cache) = get_rrd_cache() {
- if let Err(err) = rrd_cache.update_value(name, value, DST::Derive) {
- eprintln!("rrd::update_value '{}' failed - {}", name, err);
- }
+async fn generate_host_stats() {
+ match tokio::task::spawn_blocking(generate_host_stats_sync).await {
+ Ok(()) => (),
+ Err(err) => log::error!("generate_host_stats paniced: {}", err),
}
}
-async fn generate_host_stats() {
- use proxmox::sys::linux::procfs::{
+fn generate_host_stats_sync() {
+ use proxmox_sys::linux::procfs::{
read_meminfo, read_proc_stat, read_proc_net_dev, read_loadavg};
- pbs_runtime::block_in_place(move || {
-
- match read_proc_stat() {
- Ok(stat) => {
- rrd_update_gauge("host/cpu", stat.cpu);
- rrd_update_gauge("host/iowait", stat.iowait_percent);
- }
- Err(err) => {
- eprintln!("read_proc_stat failed - {}", err);
- }
+ match read_proc_stat() {
+ Ok(stat) => {
+ rrd_update_gauge("host/cpu", stat.cpu);
+ rrd_update_gauge("host/iowait", stat.iowait_percent);
+ }
+ Err(err) => {
+ eprintln!("read_proc_stat failed - {}", err);
}
+ }
- match read_meminfo() {
- Ok(meminfo) => {
- rrd_update_gauge("host/memtotal", meminfo.memtotal as f64);
- rrd_update_gauge("host/memused", meminfo.memused as f64);
- rrd_update_gauge("host/swaptotal", meminfo.swaptotal as f64);
- rrd_update_gauge("host/swapused", meminfo.swapused as f64);
- }
- Err(err) => {
- eprintln!("read_meminfo failed - {}", err);
- }
+ match read_meminfo() {
+ Ok(meminfo) => {
+ rrd_update_gauge("host/memtotal", meminfo.memtotal as f64);
+ rrd_update_gauge("host/memused", meminfo.memused as f64);
+ rrd_update_gauge("host/swaptotal", meminfo.swaptotal as f64);
+ rrd_update_gauge("host/swapused", meminfo.swapused as f64);
}
+ Err(err) => {
+ eprintln!("read_meminfo failed - {}", err);
+ }
+ }
- match read_proc_net_dev() {
- Ok(netdev) => {
- use pbs_config::network::is_physical_nic;
- let mut netin = 0;
- let mut netout = 0;
- for item in netdev {
- if !is_physical_nic(&item.device) { continue; }
- netin += item.receive;
- netout += item.send;
- }
- rrd_update_derive("host/netin", netin as f64);
- rrd_update_derive("host/netout", netout as f64);
- }
- Err(err) => {
- eprintln!("read_prox_net_dev failed - {}", err);
+ match read_proc_net_dev() {
+ Ok(netdev) => {
+ use pbs_config::network::is_physical_nic;
+ let mut netin = 0;
+ let mut netout = 0;
+ for item in netdev {
+ if !is_physical_nic(&item.device) { continue; }
+ netin += item.receive;
+ netout += item.send;
}
+ rrd_update_derive("host/netin", netin as f64);
+ rrd_update_derive("host/netout", netout as f64);
+ }
+ Err(err) => {
+ eprintln!("read_prox_net_dev failed - {}", err);
}
+ }
- match read_loadavg() {
- Ok(loadavg) => {
- rrd_update_gauge("host/loadavg", loadavg.0 as f64);
- }
- Err(err) => {
- eprintln!("read_loadavg failed - {}", err);
- }
+ match read_loadavg() {
+ Ok(loadavg) => {
+ rrd_update_gauge("host/loadavg", loadavg.0 as f64);
}
+ Err(err) => {
+ eprintln!("read_loadavg failed - {}", err);
+ }
+ }
- let disk_manager = DiskManage::new();
+ let disk_manager = DiskManage::new();
- gather_disk_stats(disk_manager.clone(), Path::new("/"), "host");
+ gather_disk_stats(disk_manager.clone(), Path::new("/"), "host");
- match pbs_config::datastore::config() {
- Ok((config, _)) => {
- let datastore_list: Vec<DataStoreConfig> =
- config.convert_to_typed_array("datastore").unwrap_or_default();
+ match pbs_config::datastore::config() {
+ Ok((config, _)) => {
+ let datastore_list: Vec<DataStoreConfig> =
+ config.convert_to_typed_array("datastore").unwrap_or_default();
- for config in datastore_list {
+ for config in datastore_list {
- let rrd_prefix = format!("datastore/{}", config.name);
- let path = std::path::Path::new(&config.path);
- gather_disk_stats(disk_manager.clone(), path, &rrd_prefix);
- }
- }
- Err(err) => {
- eprintln!("read datastore config failed - {}", err);
+ let rrd_prefix = format!("datastore/{}", config.name);
+ let path = std::path::Path::new(&config.path);
+ gather_disk_stats(disk_manager.clone(), path, &rrd_prefix);
}
}
-
- });
+ Err(err) => {
+ eprintln!("read datastore config failed - {}", err);
+ }
+ }
}
fn check_schedule(worker_type: &str, event_str: &str, id: &str) -> bool {
- let event = match parse_calendar_event(event_str) {
+ let event: CalendarEvent = match event_str.parse() {
Ok(event) => event,
Err(err) => {
eprintln!("unable to parse schedule '{}' - {}", event_str, err);
}
};
- let last = match jobstate::last_run_time(worker_type, &id) {
+ let last = match jobstate::last_run_time(worker_type, id) {
Ok(time) => time,
Err(err) => {
eprintln!("could not get last run time of {} {}: {}", worker_type, id, err);
}
};
- let next = match compute_next_event(&event, last, false) {
+ let next = match event.compute_next_event(last) {
Ok(Some(next)) => next,
Ok(None) => return false,
Err(err) => {
Ok(None) => {},
Ok(Some((fs_type, device, source))) => {
let mut device_stat = None;
- match fs_type.as_str() {
- "zfs" => {
- if let Some(source) = source {
- let pool = get_pool_from_dataset(&source).unwrap_or(&source);
- match zfs_pool_stats(pool) {
- Ok(stat) => device_stat = stat,
- Err(err) => eprintln!("zfs_pool_stats({:?}) failed - {}", pool, err),
- }
+ match (fs_type.as_str(), source) {
+ ("zfs", Some(source)) => match source.into_string() {
+ Ok(dataset) => match zfs_dataset_stats(&dataset) {
+ Ok(stat) => device_stat = Some(stat),
+ Err(err) => eprintln!("zfs_dataset_stats({:?}) failed - {}", dataset, err),
+ },
+ Err(source) => {
+ eprintln!("zfs_pool_stats({:?}) failed - invalid characters", source)
}
- }
+ },
_ => {
if let Ok(disk) = disk_manager.clone().disk_by_dev_num(device.into_dev_t()) {
match disk.read_stat() {
}
}
}
+
+// Rate Limiter lookup
+
+// Test WITH
+// proxmox-backup-client restore vm/201/2021-10-22T09:55:56Z drive-scsi0.img img1.img --repository localhost:store2
+
+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(
+ peer: Option<std::net::SocketAddr>,
+) -> (Option<Arc<dyn ShareableRateLimit>>, Option<Arc<dyn ShareableRateLimit>>) {
+ let mut cache = TRAFFIC_CONTROL_CACHE.lock().unwrap();
+
+ let now = proxmox_time::epoch_i64();
+
+ cache.reload(now);
+
+ let (_rule_name, read_limiter, write_limiter) = cache.lookup_rate_limiter(peer, now);
+
+ (read_limiter, write_limiter)
+}
projects / proxmox-backup.git / blobdiff