src/bin/proxmox-backup-proxy.rs

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