]> git.proxmox.com Git - proxmox-backup.git/blame - src/bin/proxmox-backup-proxy.rs
bump version to 2.1.4-1
[proxmox-backup.git] / src / bin / proxmox-backup-proxy.rs
CommitLineData
4ce7da51 1use std::sync::{Mutex, Arc};
2ab5acac 2use std::path::{Path, PathBuf};
97168f92 3use std::os::unix::io::AsRawFd;
6680878b
DM
4use std::future::Future;
5use std::pin::Pin;
a2479cfa 6
f7d4e4b5 7use anyhow::{bail, format_err, Error};
a2479cfa 8use futures::*;
7fa9a37c
DM
9use http::request::Parts;
10use http::Response;
11use hyper::{Body, StatusCode};
12use hyper::header;
13use url::form_urlencoded;
ea368a06 14
a2479cfa 15use openssl::ssl::{SslMethod, SslAcceptor, SslFiletype};
7c667013 16use tokio_stream::wrappers::ReceiverStream;
7fa9a37c 17use serde_json::{json, Value};
608806e8 18use http::{Method, HeaderMap};
a2479cfa 19
25877d05
DM
20use proxmox_sys::linux::socket::set_tcp_keepalive;
21use proxmox_sys::fs::CreateOptions;
6ef1b649
WB
22use proxmox_lang::try_block;
23use proxmox_router::{RpcEnvironment, RpcEnvironmentType, UserInformation};
d5f58006 24use proxmox_http::client::{RateLimitedStream, ShareableRateLimit};
d5790a9f
DM
25use proxmox_sys::{task_log, task_warn};
26use proxmox_sys::logrotate::LogRotate;
fd6d2438 27
6d5d305d 28use pbs_datastore::DataStore;
09340f28 29
48176b0a 30use proxmox_rest_server::{
608806e8 31 rotate_task_log_archive, extract_cookie , AuthError, ApiConfig, RestServer, RestEnvironment,
0e1edf19 32 ServerAdapter, WorkerTask, cleanup_old_tasks,
48176b0a 33};
a2479cfa 34
98eb435d
DM
35use proxmox_backup::rrd_cache::{
36 initialize_rrd_cache, rrd_update_gauge, rrd_update_derive, rrd_sync_journal,
37};
1298618a 38use proxmox_backup::{
a0172d76 39 TRAFFIC_CONTROL_CACHE,
1298618a 40 server::{
608806e8 41 auth::check_pbs_auth,
1298618a
DM
42 jobstate::{
43 self,
44 Job,
45 },
1298618a 46 },
1298618a
DM
47};
48
af06decd 49use pbs_buildcfg::configdir;
68b6c120 50use proxmox_time::CalendarEvent;
1298618a 51
89725197
DM
52use pbs_api_types::{
53 Authid, TapeBackupJobConfig, VerificationJobConfig, SyncJobConfig, DataStoreConfig,
54 PruneOptions,
55};
e7d4be9d 56
8bca935f
DM
57use proxmox_rest_server::daemon;
58
e3f41f21 59use proxmox_backup::server;
d01e2420 60use proxmox_backup::auth_helpers::*;
97168f92 61use proxmox_backup::tools::{
32413921 62 PROXMOX_BACKUP_TCP_KEEPALIVE_TIME,
97168f92
DM
63 disks::{
64 DiskManage,
7c069e82 65 zfs_dataset_stats,
97168f92 66 },
97168f92 67};
02c7a755 68
e7d4be9d 69
a13573c2 70use proxmox_backup::api2::pull::do_sync_job;
8513626b 71use proxmox_backup::api2::tape::backup::do_tape_backup_job;
1298618a 72use proxmox_backup::server::do_verification_job;
b8d90798 73use proxmox_backup::server::do_prune_job;
a13573c2 74
946c3e8a 75fn main() -> Result<(), Error> {
ac7513e3
DM
76 proxmox_backup::tools::setup_safe_path_env();
77
21211748
DM
78 let backup_uid = pbs_config::backup_user()?.uid;
79 let backup_gid = pbs_config::backup_group()?.gid;
843880f0
TL
80 let running_uid = nix::unistd::Uid::effective();
81 let running_gid = nix::unistd::Gid::effective();
82
83 if running_uid != backup_uid || running_gid != backup_gid {
84 bail!("proxy not running as backup user or group (got uid {} gid {})", running_uid, running_gid);
85 }
86
9a1b24b6 87 proxmox_async::runtime::main(run())
4223d9f8
DM
88}
89
48176b0a 90
608806e8
DM
91struct ProxmoxBackupProxyAdapter;
92
93impl ServerAdapter for ProxmoxBackupProxyAdapter {
94
95 fn get_index(
96 &self,
97 env: RestEnvironment,
98 parts: Parts,
99 ) -> Pin<Box<dyn Future<Output = Response<Body>> + Send>> {
100 Box::pin(get_index_future(env, parts))
101 }
102
103 fn check_auth<'a>(
104 &'a self,
105 headers: &'a HeaderMap,
106 method: &'a Method,
107 ) -> Pin<Box<dyn Future<Output = Result<(String, Box<dyn UserInformation + Sync + Send>), AuthError>> + Send + 'a>> {
108 Box::pin(async move {
109 check_pbs_auth(headers, method).await
110 })
111 }
112}
113
48176b0a
DM
114fn extract_lang_header(headers: &http::HeaderMap) -> Option<String> {
115 if let Some(Ok(cookie)) = headers.get("COOKIE").map(|v| v.to_str()) {
116 return extract_cookie(cookie, "PBSLangCookie");
117 }
118 None
119}
120
6680878b 121async fn get_index_future(
48176b0a 122 env: RestEnvironment,
7fa9a37c
DM
123 parts: Parts,
124) -> Response<Body> {
125
48176b0a
DM
126 let auth_id = env.get_auth_id();
127 let api = env.api_config();
128 let language = extract_lang_header(&parts.headers);
129
130 // fixme: make all IO async
6680878b 131
7fa9a37c
DM
132 let (userid, csrf_token) = match auth_id {
133 Some(auth_id) => {
134 let auth_id = auth_id.parse::<Authid>();
135 match auth_id {
136 Ok(auth_id) if !auth_id.is_token() => {
137 let userid = auth_id.user().clone();
138 let new_csrf_token = assemble_csrf_prevention_token(csrf_secret(), &userid);
139 (Some(userid), Some(new_csrf_token))
140 }
141 _ => (None, None)
142 }
143 }
144 None => (None, None),
145 };
146
25877d05 147 let nodename = proxmox_sys::nodename();
7fa9a37c
DM
148 let user = userid.as_ref().map(|u| u.as_str()).unwrap_or("");
149
150 let csrf_token = csrf_token.unwrap_or_else(|| String::from(""));
151
152 let mut debug = false;
153 let mut template_file = "index";
154
155 if let Some(query_str) = parts.uri.query() {
156 for (k, v) in form_urlencoded::parse(query_str.as_bytes()).into_owned() {
157 if k == "debug" && v != "0" && v != "false" {
158 debug = true;
159 } else if k == "console" {
160 template_file = "console";
161 }
162 }
163 }
164
165 let mut lang = String::from("");
166 if let Some(language) = language {
167 if Path::new(&format!("/usr/share/pbs-i18n/pbs-lang-{}.js", language)).exists() {
168 lang = language;
169 }
170 }
171
172 let data = json!({
173 "NodeName": nodename,
174 "UserName": user,
175 "CSRFPreventionToken": csrf_token,
176 "language": lang,
177 "debug": debug,
178 });
179
180 let (ct, index) = match api.render_template(template_file, &data) {
181 Ok(index) => ("text/html", index),
182 Err(err) => ("text/plain", format!("Error rendering template: {}", err)),
183 };
184
185 let mut resp = Response::builder()
186 .status(StatusCode::OK)
187 .header(header::CONTENT_TYPE, ct)
188 .body(index.into())
189 .unwrap();
190
191 if let Some(userid) = userid {
192 resp.extensions_mut().insert(Authid::from((userid, None)));
193 }
194
195 resp
196}
197
fda5797b 198async fn run() -> Result<(), Error> {
02c7a755
DM
199 if let Err(err) = syslog::init(
200 syslog::Facility::LOG_DAEMON,
201 log::LevelFilter::Info,
202 Some("proxmox-backup-proxy")) {
4223d9f8 203 bail!("unable to inititialize syslog - {}", err);
02c7a755
DM
204 }
205
e1d367df
DM
206 // Note: To debug early connection error use
207 // PROXMOX_DEBUG=1 ./target/release/proxmox-backup-proxy
208 let debug = std::env::var("PROXMOX_DEBUG").is_ok();
209
d01e2420
DM
210 let _ = public_auth_key(); // load with lazy_static
211 let _ = csrf_secret(); // load with lazy_static
212
fa49d0fd
DM
213 let rrd_cache = initialize_rrd_cache()?;
214 rrd_cache.apply_journal()?;
215
02c7a755 216 let mut config = ApiConfig::new(
af06decd 217 pbs_buildcfg::JS_DIR,
26858dba
SR
218 &proxmox_backup::api2::ROUTER,
219 RpcEnvironmentType::PUBLIC,
608806e8 220 ProxmoxBackupProxyAdapter,
26858dba 221 )?;
02c7a755 222
02c7a755
DM
223 config.add_alias("novnc", "/usr/share/novnc-pve");
224 config.add_alias("extjs", "/usr/share/javascript/extjs");
7f066a9b 225 config.add_alias("qrcodejs", "/usr/share/javascript/qrcodejs");
02c7a755
DM
226 config.add_alias("fontawesome", "/usr/share/fonts-font-awesome");
227 config.add_alias("xtermjs", "/usr/share/pve-xtermjs");
abd4c4cb 228 config.add_alias("locale", "/usr/share/pbs-i18n");
02c7a755 229 config.add_alias("widgettoolkit", "/usr/share/javascript/proxmox-widget-toolkit");
9c01e73c 230 config.add_alias("docs", "/usr/share/doc/proxmox-backup/html");
02c7a755 231
af06decd 232 let mut indexpath = PathBuf::from(pbs_buildcfg::JS_DIR);
2ab5acac
DC
233 indexpath.push("index.hbs");
234 config.register_template("index", &indexpath)?;
01ca99da 235 config.register_template("console", "/usr/share/pve-xtermjs/index.html.hbs")?;
2ab5acac 236
fd6d2438 237 let backup_user = pbs_config::backup_user()?;
49e25688 238 let mut commando_sock = proxmox_rest_server::CommandSocket::new(proxmox_rest_server::our_ctrl_sock(), backup_user.gid);
fd6d2438
DM
239
240 let dir_opts = CreateOptions::new().owner(backup_user.uid).group(backup_user.gid);
241 let file_opts = CreateOptions::new().owner(backup_user.uid).group(backup_user.gid);
a68768cf 242
0d5d15c9 243 config.enable_access_log(
fd6d2438 244 pbs_buildcfg::API_ACCESS_LOG_FN,
36b7085e
DM
245 Some(dir_opts.clone()),
246 Some(file_opts.clone()),
247 &mut commando_sock,
248 )?;
249
250 config.enable_auth_log(
251 pbs_buildcfg::API_AUTH_LOG_FN,
0a33fba4
DM
252 Some(dir_opts.clone()),
253 Some(file_opts.clone()),
fd6d2438
DM
254 &mut commando_sock,
255 )?;
8e7e2223 256
02c7a755 257 let rest_server = RestServer::new(config);
b9700a9f 258 proxmox_rest_server::init_worker_tasks(pbs_buildcfg::PROXMOX_BACKUP_LOG_DIR_M!().into(), file_opts.clone())?;
02c7a755 259
6d1f61b2 260 //openssl req -x509 -newkey rsa:4096 -keyout /etc/proxmox-backup/proxy.key -out /etc/proxmox-backup/proxy.pem -nodes
6d1f61b2 261
4ce7da51 262 // we build the initial acceptor here as we cannot start if this fails
c381a162 263 let acceptor = make_tls_acceptor()?;
4ce7da51 264 let acceptor = Arc::new(Mutex::new(acceptor));
6d1f61b2 265
4ce7da51 266 // to renew the acceptor we just add a command-socket handler
a723c087
WB
267 commando_sock.register_command(
268 "reload-certificate".to_string(),
269 {
4ce7da51 270 let acceptor = Arc::clone(&acceptor);
a723c087 271 move |_value| -> Result<_, Error> {
4ce7da51
DM
272 log::info!("reloading certificate");
273 match make_tls_acceptor() {
274 Err(err) => log::error!("error reloading certificate: {}", err),
275 Ok(new_acceptor) => {
276 let mut guard = acceptor.lock().unwrap();
277 *guard = new_acceptor;
278 }
279 }
a723c087
WB
280 Ok(Value::Null)
281 }
282 },
283 )?;
0d176f36 284
062cf75c
DC
285 // to remove references for not configured datastores
286 commando_sock.register_command(
287 "datastore-removed".to_string(),
288 |_value| {
6d5d305d 289 if let Err(err) = DataStore::remove_unused_datastores() {
062cf75c
DC
290 log::error!("could not refresh datastores: {}", err);
291 }
292 Ok(Value::Null)
293 }
294 )?;
295
a690ecac
WB
296 let server = daemon::create_daemon(
297 ([0,0,0,0,0,0,0,0], 8007).into(),
d2654200 298 move |listener| {
97168f92 299
4ce7da51 300 let connections = accept_connections(listener, acceptor, debug);
7c667013 301 let connections = hyper::server::accept::from_stream(ReceiverStream::new(connections));
083ff3fd 302
d2654200
DM
303 Ok(async {
304 daemon::systemd_notify(daemon::SystemdNotify::Ready)?;
305
306 hyper::Server::builder(connections)
083ff3fd 307 .serve(rest_server)
fd6d2438 308 .with_graceful_shutdown(proxmox_rest_server::shutdown_future())
083ff3fd 309 .map_err(Error::from)
d2654200
DM
310 .await
311 })
a2ca7137 312 },
083ff3fd 313 );
a2ca7137 314
b9700a9f 315 proxmox_rest_server::write_pid(pbs_buildcfg::PROXMOX_BACKUP_PROXY_PID_FN)?;
d98c9a7a 316
fda5797b 317 let init_result: Result<(), Error> = try_block!({
b9700a9f 318 proxmox_rest_server::register_task_control_commands(&mut commando_sock)?;
a68768cf 319 commando_sock.spawn()?;
fd1b65cc
DM
320 proxmox_rest_server::catch_shutdown_signal()?;
321 proxmox_rest_server::catch_reload_signal()?;
fda5797b
WB
322 Ok(())
323 });
d607b886 324
fda5797b
WB
325 if let Err(err) = init_result {
326 bail!("unable to start daemon - {}", err);
327 }
e3f41f21 328
8545480a 329 start_task_scheduler();
eaeda365 330 start_stat_generator();
a0172d76 331 start_traffic_control_updater();
8545480a 332
083ff3fd 333 server.await?;
a546a8a0 334 log::info!("server shutting down, waiting for active workers to complete");
fd6d2438 335 proxmox_rest_server::last_worker_future().await?;
fda5797b 336 log::info!("done - exit server");
e3f41f21 337
4223d9f8 338 Ok(())
02c7a755 339}
8545480a 340
4ce7da51 341fn make_tls_acceptor() -> Result<SslAcceptor, Error> {
c381a162
WB
342 let key_path = configdir!("/proxy.key");
343 let cert_path = configdir!("/proxy.pem");
344
2eba3967 345 let (config, _) = proxmox_backup::config::node::config()?;
5ee8dd78
FG
346 let ciphers_tls_1_3 = config.ciphers_tls_1_3;
347 let ciphers_tls_1_2 = config.ciphers_tls_1_2;
2eba3967 348
c381a162 349 let mut acceptor = SslAcceptor::mozilla_intermediate_v5(SslMethod::tls()).unwrap();
5ee8dd78 350 if let Some(ciphers) = ciphers_tls_1_3.as_deref() {
2eba3967
HL
351 acceptor.set_ciphersuites(ciphers)?;
352 }
5ee8dd78 353 if let Some(ciphers) = ciphers_tls_1_2.as_deref() {
2eba3967
HL
354 acceptor.set_cipher_list(ciphers)?;
355 }
c381a162
WB
356 acceptor.set_private_key_file(key_path, SslFiletype::PEM)
357 .map_err(|err| format_err!("unable to read proxy key {} - {}", key_path, err))?;
358 acceptor.set_certificate_chain_file(cert_path)
359 .map_err(|err| format_err!("unable to read proxy cert {} - {}", cert_path, err))?;
a0c69902 360 acceptor.set_options(openssl::ssl::SslOptions::NO_RENEGOTIATION);
c381a162
WB
361 acceptor.check_private_key().unwrap();
362
4ce7da51 363 Ok(acceptor.build())
c381a162
WB
364}
365
a5e3be49 366type ClientStreamResult =
e511e0e5 367 Result<std::pin::Pin<Box<tokio_openssl::SslStream<RateLimitedStream<tokio::net::TcpStream>>>>, Error>;
a5e3be49
WB
368const MAX_PENDING_ACCEPTS: usize = 1024;
369
48aa2b93 370fn accept_connections(
0bfcea6a 371 listener: tokio::net::TcpListener,
4ce7da51 372 acceptor: Arc<Mutex<openssl::ssl::SslAcceptor>>,
e1d367df 373 debug: bool,
a5e3be49 374) -> tokio::sync::mpsc::Receiver<ClientStreamResult> {
48aa2b93 375
ea93bea7 376 let (sender, receiver) = tokio::sync::mpsc::channel(MAX_PENDING_ACCEPTS);
48aa2b93 377
4ce7da51 378 tokio::spawn(accept_connection(listener, acceptor, debug, sender));
a5e3be49
WB
379
380 receiver
381}
382
383async fn accept_connection(
384 listener: tokio::net::TcpListener,
4ce7da51 385 acceptor: Arc<Mutex<openssl::ssl::SslAcceptor>>,
a5e3be49
WB
386 debug: bool,
387 sender: tokio::sync::mpsc::Sender<ClientStreamResult>,
388) {
ea93bea7 389 let accept_counter = Arc::new(());
48aa2b93 390
a5e3be49 391 loop {
4ce7da51
DM
392 let (sock, _addr) = match listener.accept().await {
393 Ok(conn) => conn,
394 Err(err) => {
395 eprintln!("error accepting tcp connection: {}", err);
cc269b9f 396 continue;
a5e3be49 397 }
cc269b9f 398 };
48aa2b93 399
cc269b9f
WB
400 sock.set_nodelay(true).unwrap();
401 let _ = set_tcp_keepalive(sock.as_raw_fd(), PROXMOX_BACKUP_TCP_KEEPALIVE_TIME);
48aa2b93 402
e511e0e5
DM
403 let peer = sock.peer_addr().ok();
404 let sock = RateLimitedStream::with_limiter_update_cb(sock, move || lookup_rate_limiter(peer));
405
4ce7da51
DM
406 let ssl = { // limit acceptor_guard scope
407 // Acceptor can be reloaded using the command socket "reload-certificate" command
408 let acceptor_guard = acceptor.lock().unwrap();
409
410 match openssl::ssl::Ssl::new(acceptor_guard.context()) {
411 Ok(ssl) => ssl,
412 Err(err) => {
413 eprintln!("failed to create Ssl object from Acceptor context - {}", err);
414 continue;
415 },
416 }
cc269b9f 417 };
4ce7da51 418
cc269b9f
WB
419 let stream = match tokio_openssl::SslStream::new(ssl, sock) {
420 Ok(stream) => stream,
421 Err(err) => {
422 eprintln!("failed to create SslStream using ssl and connection socket - {}", err);
423 continue;
424 },
425 };
426
427 let mut stream = Box::pin(stream);
428 let sender = sender.clone();
429
430 if Arc::strong_count(&accept_counter) > MAX_PENDING_ACCEPTS {
431 eprintln!("connection rejected - to many open connections");
432 continue;
48aa2b93 433 }
cc269b9f 434
b4931192 435 let accept_counter = Arc::clone(&accept_counter);
cc269b9f
WB
436 tokio::spawn(async move {
437 let accept_future = tokio::time::timeout(
438 Duration::new(10, 0), stream.as_mut().accept());
439
440 let result = accept_future.await;
441
442 match result {
443 Ok(Ok(())) => {
444 if sender.send(Ok(stream)).await.is_err() && debug {
445 eprintln!("detect closed connection channel");
446 }
447 }
448 Ok(Err(err)) => {
449 if debug {
450 eprintln!("https handshake failed - {}", err);
451 }
452 }
453 Err(_) => {
454 if debug {
455 eprintln!("https handshake timeout");
456 }
457 }
458 }
459
460 drop(accept_counter); // decrease reference count
461 });
a5e3be49 462 }
48aa2b93
DM
463}
464
eaeda365 465fn start_stat_generator() {
fd6d2438 466 let abort_future = proxmox_rest_server::shutdown_future();
eaeda365
DM
467 let future = Box::pin(run_stat_generator());
468 let task = futures::future::select(future, abort_future);
469 tokio::spawn(task.map(|_| ()));
470}
471
8545480a 472fn start_task_scheduler() {
fd6d2438 473 let abort_future = proxmox_rest_server::shutdown_future();
8545480a
DM
474 let future = Box::pin(run_task_scheduler());
475 let task = futures::future::select(future, abort_future);
476 tokio::spawn(task.map(|_| ()));
477}
478
a0172d76
DM
479fn start_traffic_control_updater() {
480 let abort_future = proxmox_rest_server::shutdown_future();
481 let future = Box::pin(run_traffic_control_updater());
482 let task = futures::future::select(future, abort_future);
483 tokio::spawn(task.map(|_| ()));
484}
485
6a7be83e 486use std::time::{SystemTime, Instant, Duration, UNIX_EPOCH};
8545480a
DM
487
488fn next_minute() -> Result<Instant, Error> {
6a7be83e
DM
489 let now = SystemTime::now();
490 let epoch_now = now.duration_since(UNIX_EPOCH)?;
491 let epoch_next = Duration::from_secs((epoch_now.as_secs()/60 + 1)*60);
8545480a
DM
492 Ok(Instant::now() + epoch_next - epoch_now)
493}
494
495async fn run_task_scheduler() {
496
497 let mut count: usize = 0;
498
499 loop {
500 count += 1;
501
502 let delay_target = match next_minute() { // try to run very minute
503 Ok(d) => d,
504 Err(err) => {
505 eprintln!("task scheduler: compute next minute failed - {}", err);
0a8d773a 506 tokio::time::sleep_until(tokio::time::Instant::from_std(Instant::now() + Duration::from_secs(60))).await;
8545480a
DM
507 continue;
508 }
509 };
510
511 if count > 2 { // wait 1..2 minutes before starting
512 match schedule_tasks().catch_unwind().await {
513 Err(panic) => {
514 match panic.downcast::<&str>() {
515 Ok(msg) => {
516 eprintln!("task scheduler panic: {}", msg);
517 }
518 Err(_) => {
519 eprintln!("task scheduler panic - unknown type");
520 }
521 }
522 }
523 Ok(Err(err)) => {
524 eprintln!("task scheduler failed - {:?}", err);
525 }
526 Ok(Ok(_)) => {}
527 }
528 }
529
0a8d773a 530 tokio::time::sleep_until(tokio::time::Instant::from_std(delay_target)).await;
8545480a
DM
531 }
532}
533
534async fn schedule_tasks() -> Result<(), Error> {
535
536 schedule_datastore_garbage_collection().await;
25829a87 537 schedule_datastore_prune().await;
a6160cdf 538 schedule_datastore_sync_jobs().await;
73df9c51 539 schedule_datastore_verify_jobs().await;
8513626b 540 schedule_tape_backup_jobs().await;
9a760917 541 schedule_task_log_rotate().await;
8545480a
DM
542
543 Ok(())
544}
545
8545480a
DM
546async fn schedule_datastore_garbage_collection() {
547
e7d4be9d 548 let config = match pbs_config::datastore::config() {
8545480a
DM
549 Err(err) => {
550 eprintln!("unable to read datastore config - {}", err);
551 return;
552 }
553 Ok((config, _digest)) => config,
554 };
555
556 for (store, (_, store_config)) in config.sections {
557 let datastore = match DataStore::lookup_datastore(&store) {
558 Ok(datastore) => datastore,
559 Err(err) => {
560 eprintln!("lookup_datastore failed - {}", err);
561 continue;
562 }
563 };
564
25829a87 565 let store_config: DataStoreConfig = match serde_json::from_value(store_config) {
8545480a
DM
566 Ok(c) => c,
567 Err(err) => {
568 eprintln!("datastore config from_value failed - {}", err);
569 continue;
570 }
571 };
572
573 let event_str = match store_config.gc_schedule {
574 Some(event_str) => event_str,
575 None => continue,
576 };
577
68b6c120 578 let event: CalendarEvent = match event_str.parse() {
8545480a
DM
579 Ok(event) => event,
580 Err(err) => {
581 eprintln!("unable to parse schedule '{}' - {}", event_str, err);
582 continue;
583 }
584 };
585
586 if datastore.garbage_collection_running() { continue; }
587
588 let worker_type = "garbage_collection";
589
b6ba5acd
DC
590 let last = match jobstate::last_run_time(worker_type, &store) {
591 Ok(time) => time,
592 Err(err) => {
593 eprintln!("could not get last run time of {} {}: {}", worker_type, store, err);
594 continue;
8545480a
DM
595 }
596 };
597
7549114c 598 let next = match event.compute_next_event(last) {
15ec790a
DC
599 Ok(Some(next)) => next,
600 Ok(None) => continue,
8545480a
DM
601 Err(err) => {
602 eprintln!("compute_next_event for '{}' failed - {}", event_str, err);
603 continue;
604 }
605 };
e693818a 606
6ef1b649 607 let now = proxmox_time::epoch_i64();
6a7be83e 608
8545480a
DM
609 if next > now { continue; }
610
1cd951c9 611 let job = match Job::new(worker_type, &store) {
d7a122a0
DC
612 Ok(job) => job,
613 Err(_) => continue, // could not get lock
614 };
615
ad54df31 616 let auth_id = Authid::root_auth_id();
d7a122a0 617
c724f658 618 if let Err(err) = crate::server::do_garbage_collection_job(job, datastore, auth_id, Some(event_str), false) {
3b707fbb 619 eprintln!("unable to start garbage collection job on datastore {} - {}", store, err);
8545480a
DM
620 }
621 }
622}
25829a87
DM
623
624async fn schedule_datastore_prune() {
625
e7d4be9d 626 let config = match pbs_config::datastore::config() {
25829a87
DM
627 Err(err) => {
628 eprintln!("unable to read datastore config - {}", err);
629 return;
630 }
631 Ok((config, _digest)) => config,
632 };
633
634 for (store, (_, store_config)) in config.sections {
25829a87
DM
635
636 let store_config: DataStoreConfig = match serde_json::from_value(store_config) {
637 Ok(c) => c,
638 Err(err) => {
a6160cdf 639 eprintln!("datastore '{}' config from_value failed - {}", store, err);
25829a87
DM
640 continue;
641 }
642 };
643
644 let event_str = match store_config.prune_schedule {
645 Some(event_str) => event_str,
646 None => continue,
647 };
648
649 let prune_options = PruneOptions {
650 keep_last: store_config.keep_last,
651 keep_hourly: store_config.keep_hourly,
652 keep_daily: store_config.keep_daily,
653 keep_weekly: store_config.keep_weekly,
654 keep_monthly: store_config.keep_monthly,
655 keep_yearly: store_config.keep_yearly,
656 };
657
89725197 658 if !pbs_datastore::prune::keeps_something(&prune_options) { // no prune settings - keep all
25829a87
DM
659 continue;
660 }
661
25829a87 662 let worker_type = "prune";
b15751bf 663 if check_schedule(worker_type, &event_str, &store) {
82c05b41
HL
664 let job = match Job::new(worker_type, &store) {
665 Ok(job) => job,
666 Err(_) => continue, // could not get lock
667 };
25829a87 668
ad54df31 669 let auth_id = Authid::root_auth_id().clone();
82c05b41
HL
670 if let Err(err) = do_prune_job(job, prune_options, store.clone(), &auth_id, Some(event_str)) {
671 eprintln!("unable to start datastore prune job {} - {}", &store, err);
25829a87
DM
672 }
673 };
25829a87
DM
674 }
675}
a6160cdf
DM
676
677async fn schedule_datastore_sync_jobs() {
678
a6160cdf 679
a4e5a0fc 680 let config = match pbs_config::sync::config() {
a6160cdf
DM
681 Err(err) => {
682 eprintln!("unable to read sync job config - {}", err);
683 return;
684 }
685 Ok((config, _digest)) => config,
686 };
687
a6160cdf
DM
688 for (job_id, (_, job_config)) in config.sections {
689 let job_config: SyncJobConfig = match serde_json::from_value(job_config) {
690 Ok(c) => c,
691 Err(err) => {
692 eprintln!("sync job config from_value failed - {}", err);
693 continue;
694 }
695 };
696
697 let event_str = match job_config.schedule {
698 Some(ref event_str) => event_str.clone(),
699 None => continue,
700 };
701
c67b1fa7 702 let worker_type = "syncjob";
b15751bf 703 if check_schedule(worker_type, &event_str, &job_id) {
82c05b41
HL
704 let job = match Job::new(worker_type, &job_id) {
705 Ok(job) => job,
706 Err(_) => continue, // could not get lock
707 };
a6160cdf 708
ad54df31 709 let auth_id = Authid::root_auth_id().clone();
bfa942c0 710 if let Err(err) = do_sync_job(job, job_config, &auth_id, Some(event_str), false) {
82c05b41 711 eprintln!("unable to start datastore sync job {} - {}", &job_id, err);
a6160cdf
DM
712 }
713 };
a6160cdf
DM
714 }
715}
eaeda365 716
73df9c51 717async fn schedule_datastore_verify_jobs() {
1298618a 718
802189f7 719 let config = match pbs_config::verify::config() {
73df9c51
HL
720 Err(err) => {
721 eprintln!("unable to read verification job config - {}", err);
722 return;
723 }
724 Ok((config, _digest)) => config,
725 };
726 for (job_id, (_, job_config)) in config.sections {
727 let job_config: VerificationJobConfig = match serde_json::from_value(job_config) {
728 Ok(c) => c,
729 Err(err) => {
730 eprintln!("verification job config from_value failed - {}", err);
731 continue;
732 }
733 };
734 let event_str = match job_config.schedule {
735 Some(ref event_str) => event_str.clone(),
736 None => continue,
737 };
82c05b41 738
73df9c51 739 let worker_type = "verificationjob";
ad54df31 740 let auth_id = Authid::root_auth_id().clone();
b15751bf 741 if check_schedule(worker_type, &event_str, &job_id) {
9a37bd6c 742 let job = match Job::new(worker_type, &job_id) {
82c05b41
HL
743 Ok(job) => job,
744 Err(_) => continue, // could not get lock
745 };
bfa942c0 746 if let Err(err) = do_verification_job(job, job_config, &auth_id, Some(event_str), false) {
82c05b41 747 eprintln!("unable to start datastore verification job {} - {}", &job_id, err);
73df9c51
HL
748 }
749 };
73df9c51
HL
750 }
751}
752
8513626b
DM
753async fn schedule_tape_backup_jobs() {
754
e3619d41 755 let config = match pbs_config::tape_job::config() {
8513626b
DM
756 Err(err) => {
757 eprintln!("unable to read tape job config - {}", err);
758 return;
759 }
760 Ok((config, _digest)) => config,
761 };
762 for (job_id, (_, job_config)) in config.sections {
763 let job_config: TapeBackupJobConfig = match serde_json::from_value(job_config) {
764 Ok(c) => c,
765 Err(err) => {
766 eprintln!("tape backup job config from_value failed - {}", err);
767 continue;
768 }
769 };
770 let event_str = match job_config.schedule {
771 Some(ref event_str) => event_str.clone(),
772 None => continue,
773 };
774
775 let worker_type = "tape-backup-job";
776 let auth_id = Authid::root_auth_id().clone();
777 if check_schedule(worker_type, &event_str, &job_id) {
9a37bd6c 778 let job = match Job::new(worker_type, &job_id) {
8513626b
DM
779 Ok(job) => job,
780 Err(_) => continue, // could not get lock
781 };
bfa942c0 782 if let Err(err) = do_tape_backup_job(job, job_config.setup, &auth_id, Some(event_str), false) {
7a61f89e 783 eprintln!("unable to start tape backup job {} - {}", &job_id, err);
8513626b
DM
784 }
785 };
786 }
787}
788
789
9a760917 790async fn schedule_task_log_rotate() {
9a760917
DC
791
792 let worker_type = "logrotate";
72aa1834 793 let job_id = "access-log_and_task-archive";
9a760917 794
9a760917
DC
795 // schedule daily at 00:00 like normal logrotate
796 let schedule = "00:00";
797
b15751bf 798 if !check_schedule(worker_type, schedule, job_id) {
9a760917
DC
799 // if we never ran the rotation, schedule instantly
800 match jobstate::JobState::load(worker_type, job_id) {
801 Ok(state) => match state {
802 jobstate::JobState::Created { .. } => {},
803 _ => return,
804 },
805 _ => return,
806 }
807 }
808
809 let mut job = match Job::new(worker_type, job_id) {
810 Ok(job) => job,
811 Err(_) => return, // could not get lock
812 };
813
814 if let Err(err) = WorkerTask::new_thread(
815 worker_type,
72aa1834 816 None,
049a22a3 817 Authid::root_auth_id().to_string(),
9a760917
DC
818 false,
819 move |worker| {
820 job.start(&worker.upid().to_string())?;
1ec0d70d 821 task_log!(worker, "starting task log rotation");
e4f5f59e 822
9a760917 823 let result = try_block!({
b7f2be51
TL
824 let max_size = 512 * 1024 - 1; // an entry has ~ 100b, so > 5000 entries/file
825 let max_files = 20; // times twenty files gives > 100000 task entries
d5790a9f
DM
826
827 let user = pbs_config::backup_user()?;
25877d05 828 let options = proxmox_sys::fs::CreateOptions::new()
d5790a9f
DM
829 .owner(user.uid)
830 .group(user.gid);
831
832 let has_rotated = rotate_task_log_archive(
833 max_size,
834 true,
835 Some(max_files),
836 Some(options.clone()),
837 )?;
838
9a760917 839 if has_rotated {
1ec0d70d 840 task_log!(worker, "task log archive was rotated");
9a760917 841 } else {
1ec0d70d 842 task_log!(worker, "task log archive was not rotated");
9a760917
DC
843 }
844
fe4cc5b1
TL
845 let max_size = 32 * 1024 * 1024 - 1;
846 let max_files = 14;
fe4cc5b1 847
d5790a9f
DM
848
849 let mut logrotate = LogRotate::new(
850 pbs_buildcfg::API_ACCESS_LOG_FN,
851 true,
852 Some(max_files),
853 Some(options.clone()),
854 )?;
855
856 if logrotate.rotate(max_size)? {
fe4cc5b1 857 println!("rotated access log, telling daemons to re-open log file");
9a1b24b6 858 proxmox_async::runtime::block_on(command_reopen_access_logfiles())?;
1ec0d70d 859 task_log!(worker, "API access log was rotated");
fe7bdc9d 860 } else {
1ec0d70d 861 task_log!(worker, "API access log was not rotated");
fe7bdc9d
TL
862 }
863
d5790a9f
DM
864 let mut logrotate = LogRotate::new(
865 pbs_buildcfg::API_AUTH_LOG_FN,
866 true,
867 Some(max_files),
868 Some(options),
869 )?;
fe4cc5b1 870
d5790a9f 871 if logrotate.rotate(max_size)? {
36b7085e 872 println!("rotated auth log, telling daemons to re-open log file");
9a1b24b6 873 proxmox_async::runtime::block_on(command_reopen_auth_logfiles())?;
1ec0d70d 874 task_log!(worker, "API authentication log was rotated");
fe4cc5b1 875 } else {
1ec0d70d 876 task_log!(worker, "API authentication log was not rotated");
fe4cc5b1
TL
877 }
878
0e1edf19
DC
879 if has_rotated {
880 task_log!(worker, "cleaning up old task logs");
881 if let Err(err) = cleanup_old_tasks(true) {
882 task_warn!(worker, "could not completely cleanup old tasks: {}", err);
883 }
884 }
885
9a760917
DC
886 Ok(())
887 });
888
889 let status = worker.create_state(&result);
890
891 if let Err(err) = job.finish(status) {
892 eprintln!("could not finish job state for {}: {}", worker_type, err);
893 }
894
895 result
896 },
897 ) {
898 eprintln!("unable to start task log rotation: {}", err);
899 }
900
901}
902
36b7085e 903async fn command_reopen_access_logfiles() -> Result<(), Error> {
fe4cc5b1
TL
904 // only care about the most recent daemon instance for each, proxy & api, as other older ones
905 // should not respond to new requests anyway, but only finish their current one and then exit.
b9700a9f 906 let sock = proxmox_rest_server::our_ctrl_sock();
75442e81 907 let f1 = proxmox_rest_server::send_raw_command(sock, "{\"command\":\"api-access-log-reopen\"}\n");
fe4cc5b1 908
b9700a9f
DM
909 let pid = proxmox_rest_server::read_pid(pbs_buildcfg::PROXMOX_BACKUP_API_PID_FN)?;
910 let sock = proxmox_rest_server::ctrl_sock_from_pid(pid);
75442e81 911 let f2 = proxmox_rest_server::send_raw_command(sock, "{\"command\":\"api-access-log-reopen\"}\n");
546b6a23
TL
912
913 match futures::join!(f1, f2) {
914 (Err(e1), Err(e2)) => Err(format_err!("reopen commands failed, proxy: {}; api: {}", e1, e2)),
915 (Err(e1), Ok(_)) => Err(format_err!("reopen commands failed, proxy: {}", e1)),
36b7085e
DM
916 (Ok(_), Err(e2)) => Err(format_err!("reopen commands failed, api: {}", e2)),
917 _ => Ok(()),
918 }
919}
920
921async fn command_reopen_auth_logfiles() -> Result<(), Error> {
922 // only care about the most recent daemon instance for each, proxy & api, as other older ones
923 // should not respond to new requests anyway, but only finish their current one and then exit.
b9700a9f 924 let sock = proxmox_rest_server::our_ctrl_sock();
75442e81 925 let f1 = proxmox_rest_server::send_raw_command(sock, "{\"command\":\"api-auth-log-reopen\"}\n");
36b7085e 926
b9700a9f
DM
927 let pid = proxmox_rest_server::read_pid(pbs_buildcfg::PROXMOX_BACKUP_API_PID_FN)?;
928 let sock = proxmox_rest_server::ctrl_sock_from_pid(pid);
75442e81 929 let f2 = proxmox_rest_server::send_raw_command(sock, "{\"command\":\"api-auth-log-reopen\"}\n");
36b7085e
DM
930
931 match futures::join!(f1, f2) {
932 (Err(e1), Err(e2)) => Err(format_err!("reopen commands failed, proxy: {}; api: {}", e1, e2)),
933 (Err(e1), Ok(_)) => Err(format_err!("reopen commands failed, proxy: {}", e1)),
546b6a23
TL
934 (Ok(_), Err(e2)) => Err(format_err!("reopen commands failed, api: {}", e2)),
935 _ => Ok(()),
936 }
fe4cc5b1
TL
937}
938
eaeda365
DM
939async fn run_stat_generator() {
940
941 loop {
942 let delay_target = Instant::now() + Duration::from_secs(10);
943
1d44f175 944 generate_host_stats().await;
eaeda365 945
98eb435d
DM
946 rrd_sync_journal();
947
0a8d773a 948 tokio::time::sleep_until(tokio::time::Instant::from_std(delay_target)).await;
013fa7bb
DM
949
950 }
eaeda365
DM
951
952}
953
1d44f175 954async fn generate_host_stats() {
4b709ade
DM
955 match tokio::task::spawn_blocking(generate_host_stats_sync).await {
956 Ok(()) => (),
957 Err(err) => log::error!("generate_host_stats paniced: {}", err),
958 }
959}
960
961fn generate_host_stats_sync() {
25877d05 962 use proxmox_sys::linux::procfs::{
485841da 963 read_meminfo, read_proc_stat, read_proc_net_dev, read_loadavg};
eaeda365 964
4b709ade
DM
965 match read_proc_stat() {
966 Ok(stat) => {
967 rrd_update_gauge("host/cpu", stat.cpu);
968 rrd_update_gauge("host/iowait", stat.iowait_percent);
eaeda365 969 }
4b709ade
DM
970 Err(err) => {
971 eprintln!("read_proc_stat failed - {}", err);
972 }
973 }
2c66a590 974
4b709ade
DM
975 match read_meminfo() {
976 Ok(meminfo) => {
977 rrd_update_gauge("host/memtotal", meminfo.memtotal as f64);
978 rrd_update_gauge("host/memused", meminfo.memused as f64);
979 rrd_update_gauge("host/swaptotal", meminfo.swaptotal as f64);
980 rrd_update_gauge("host/swapused", meminfo.swapused as f64);
981 }
982 Err(err) => {
983 eprintln!("read_meminfo failed - {}", err);
a4a3f7ca 984 }
4b709ade 985 }
8f0cec26 986
4b709ade
DM
987 match read_proc_net_dev() {
988 Ok(netdev) => {
989 use pbs_config::network::is_physical_nic;
990 let mut netin = 0;
991 let mut netout = 0;
992 for item in netdev {
993 if !is_physical_nic(&item.device) { continue; }
994 netin += item.receive;
995 netout += item.send;
996 }
997 rrd_update_derive("host/netin", netin as f64);
998 rrd_update_derive("host/netout", netout as f64);
999 }
1000 Err(err) => {
1001 eprintln!("read_prox_net_dev failed - {}", err);
8f0cec26 1002 }
4b709ade 1003 }
dd15c0aa 1004
4b709ade
DM
1005 match read_loadavg() {
1006 Ok(loadavg) => {
1007 rrd_update_gauge("host/loadavg", loadavg.0 as f64);
1008 }
1009 Err(err) => {
1010 eprintln!("read_loadavg failed - {}", err);
485841da 1011 }
4b709ade 1012 }
485841da 1013
4b709ade 1014 let disk_manager = DiskManage::new();
8c03041a 1015
4b709ade 1016 gather_disk_stats(disk_manager.clone(), Path::new("/"), "host");
91e5bb49 1017
4b709ade
DM
1018 match pbs_config::datastore::config() {
1019 Ok((config, _)) => {
1020 let datastore_list: Vec<DataStoreConfig> =
1021 config.convert_to_typed_array("datastore").unwrap_or_default();
d0833a70 1022
4b709ade 1023 for config in datastore_list {
8c03041a 1024
4b709ade
DM
1025 let rrd_prefix = format!("datastore/{}", config.name);
1026 let path = std::path::Path::new(&config.path);
1027 gather_disk_stats(disk_manager.clone(), path, &rrd_prefix);
d0833a70
DM
1028 }
1029 }
4b709ade
DM
1030 Err(err) => {
1031 eprintln!("read datastore config failed - {}", err);
1032 }
1033 }
eaeda365 1034}
dd15c0aa 1035
b15751bf 1036fn check_schedule(worker_type: &str, event_str: &str, id: &str) -> bool {
68b6c120 1037 let event: CalendarEvent = match event_str.parse() {
82c05b41
HL
1038 Ok(event) => event,
1039 Err(err) => {
1040 eprintln!("unable to parse schedule '{}' - {}", event_str, err);
1041 return false;
1042 }
1043 };
1044
9a37bd6c 1045 let last = match jobstate::last_run_time(worker_type, id) {
82c05b41
HL
1046 Ok(time) => time,
1047 Err(err) => {
1048 eprintln!("could not get last run time of {} {}: {}", worker_type, id, err);
1049 return false;
1050 }
1051 };
1052
7549114c 1053 let next = match event.compute_next_event(last) {
82c05b41
HL
1054 Ok(Some(next)) => next,
1055 Ok(None) => return false,
1056 Err(err) => {
1057 eprintln!("compute_next_event for '{}' failed - {}", event_str, err);
1058 return false;
1059 }
1060 };
1061
6ef1b649 1062 let now = proxmox_time::epoch_i64();
82c05b41
HL
1063 next <= now
1064}
1065
1d44f175 1066fn gather_disk_stats(disk_manager: Arc<DiskManage>, path: &Path, rrd_prefix: &str) {
91e5bb49 1067
934f5bb8 1068 match proxmox_backup::tools::disks::disk_usage(path) {
33070956 1069 Ok(status) => {
91e5bb49 1070 let rrd_key = format!("{}/total", rrd_prefix);
1d44f175 1071 rrd_update_gauge(&rrd_key, status.total as f64);
91e5bb49 1072 let rrd_key = format!("{}/used", rrd_prefix);
1d44f175 1073 rrd_update_gauge(&rrd_key, status.used as f64);
91e5bb49
DM
1074 }
1075 Err(err) => {
1076 eprintln!("read disk_usage on {:?} failed - {}", path, err);
1077 }
1078 }
1079
934f5bb8
DM
1080 match disk_manager.find_mounted_device(path) {
1081 Ok(None) => {},
1082 Ok(Some((fs_type, device, source))) => {
1083 let mut device_stat = None;
7c069e82
DC
1084 match (fs_type.as_str(), source) {
1085 ("zfs", Some(source)) => match source.into_string() {
1086 Ok(dataset) => match zfs_dataset_stats(&dataset) {
1087 Ok(stat) => device_stat = Some(stat),
1088 Err(err) => eprintln!("zfs_dataset_stats({:?}) failed - {}", dataset, err),
1089 },
1090 Err(source) => {
1091 eprintln!("zfs_pool_stats({:?}) failed - invalid characters", source)
91e5bb49 1092 }
7c069e82 1093 },
934f5bb8
DM
1094 _ => {
1095 if let Ok(disk) = disk_manager.clone().disk_by_dev_num(device.into_dev_t()) {
1096 match disk.read_stat() {
1097 Ok(stat) => device_stat = stat,
1098 Err(err) => eprintln!("disk.read_stat {:?} failed - {}", path, err),
91e5bb49
DM
1099 }
1100 }
1101 }
91e5bb49 1102 }
934f5bb8
DM
1103 if let Some(stat) = device_stat {
1104 let rrd_key = format!("{}/read_ios", rrd_prefix);
1d44f175 1105 rrd_update_derive(&rrd_key, stat.read_ios as f64);
934f5bb8 1106 let rrd_key = format!("{}/read_bytes", rrd_prefix);
1d44f175 1107 rrd_update_derive(&rrd_key, (stat.read_sectors*512) as f64);
dd15c0aa 1108
934f5bb8 1109 let rrd_key = format!("{}/write_ios", rrd_prefix);
1d44f175 1110 rrd_update_derive(&rrd_key, stat.write_ios as f64);
934f5bb8 1111 let rrd_key = format!("{}/write_bytes", rrd_prefix);
1d44f175 1112 rrd_update_derive(&rrd_key, (stat.write_sectors*512) as f64);
dd15c0aa 1113
934f5bb8 1114 let rrd_key = format!("{}/io_ticks", rrd_prefix);
1d44f175 1115 rrd_update_derive(&rrd_key, (stat.io_ticks as f64)/1000.0);
8c03041a
DM
1116 }
1117 }
934f5bb8
DM
1118 Err(err) => {
1119 eprintln!("find_mounted_device failed - {}", err);
1120 }
8c03041a 1121 }
8c03041a 1122}
e511e0e5
DM
1123
1124// Rate Limiter lookup
1125
1126// Test WITH
1127// proxmox-backup-client restore vm/201/2021-10-22T09:55:56Z drive-scsi0.img img1.img --repository localhost:store2
1128
a0172d76
DM
1129async fn run_traffic_control_updater() {
1130
1131 loop {
1132 let delay_target = Instant::now() + Duration::from_secs(1);
1133
1134 {
1135 let mut cache = TRAFFIC_CONTROL_CACHE.lock().unwrap();
1136 cache.compute_current_rates();
1137 }
1138
1139 tokio::time::sleep_until(tokio::time::Instant::from_std(delay_target)).await;
1140 }
1141
e511e0e5
DM
1142}
1143
1144fn lookup_rate_limiter(
1145 peer: Option<std::net::SocketAddr>,
d5f58006 1146) -> (Option<Arc<dyn ShareableRateLimit>>, Option<Arc<dyn ShareableRateLimit>>) {
e511e0e5
DM
1147 let mut cache = TRAFFIC_CONTROL_CACHE.lock().unwrap();
1148
1149 let now = proxmox_time::epoch_i64();
1150
1151 cache.reload(now);
1152
1153 let (_rule_name, read_limiter, write_limiter) = cache.lookup_rate_limiter(peer, now);
1154
1155 (read_limiter, write_limiter)
1156}