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
   5 use anyhow::{bail, format_err, Error};
   6 use futures::*;
   7
   8 use openssl::ssl::{SslMethod, SslAcceptor, SslFiletype};
   9 use tokio_stream::wrappers::ReceiverStream;
  10 use serde_json::Value;
  11
  12 use proxmox::try_block;
  13 use proxmox::api::RpcEnvironmentType;
  14 use proxmox::sys::linux::socket::set_tcp_keepalive;
  15 use proxmox::tools::fs::CreateOptions;
  16
  17 use proxmox_rest_server::ApiConfig;
  18
  19 use proxmox_backup::{
  20     backup::DataStore,
  21     server::{
  22         auth::default_api_auth,
  23         WorkerTask,
  24         rest::*,
  25         jobstate::{
  26             self,
  27             Job,
  28         },
  29         rotate_task_log_archive,
  30     },
  31 };
  32
  33 use pbs_buildcfg::configdir;
  34 use pbs_systemd::time::{compute_next_event, parse_calendar_event};
  35 use pbs_tools::logrotate::LogRotate;
  36
  37 use pbs_api_types::{
  38     Authid, TapeBackupJobConfig, VerificationJobConfig, SyncJobConfig, DataStoreConfig,
  39     PruneOptions,
  40 };
  41
  42 use proxmox_rest_server::daemon;
  43
  44 use proxmox_backup::server;
  45 use proxmox_backup::auth_helpers::*;
  46 use proxmox_backup::tools::{
  47     PROXMOX_BACKUP_TCP_KEEPALIVE_TIME,
  48     disks::{
  49         DiskManage,
  50         zfs_pool_stats,
  51         get_pool_from_dataset,
  52     },
  53 };
  54
  55
  56 use proxmox_backup::api2::pull::do_sync_job;
  57 use proxmox_backup::api2::tape::backup::do_tape_backup_job;
  58 use proxmox_backup::server::do_verification_job;
  59 use proxmox_backup::server::do_prune_job;
  60
  61 fn main() -> Result<(), Error> {
  62     proxmox_backup::tools::setup_safe_path_env();
  63
  64     let backup_uid = pbs_config::backup_user()?.uid;
  65     let backup_gid = pbs_config::backup_group()?.gid;
  66     let running_uid = nix::unistd::Uid::effective();
  67     let running_gid = nix::unistd::Gid::effective();
  68
  69     if running_uid != backup_uid || running_gid != backup_gid {
  70         bail!("proxy not running as backup user or group (got uid {} gid {})", running_uid, running_gid);
  71     }
  72
  73     pbs_runtime::main(run())
  74 }
  75
  76 async fn run() -> Result<(), Error> {
  77     if let Err(err) = syslog::init(
  78         syslog::Facility::LOG_DAEMON,
  79         log::LevelFilter::Info,
  80         Some("proxmox-backup-proxy")) {
  81         bail!("unable to inititialize syslog - {}", err);
  82     }
  83
  84     // Note: To debug early connection error use
  85     // PROXMOX_DEBUG=1 ./target/release/proxmox-backup-proxy
  86     let debug = std::env::var("PROXMOX_DEBUG").is_ok();
  87
  88     let _ = public_auth_key(); // load with lazy_static
  89     let _ = csrf_secret(); // load with lazy_static
  90
  91     let mut config = ApiConfig::new(
  92         pbs_buildcfg::JS_DIR,
  93         &proxmox_backup::api2::ROUTER,
  94         RpcEnvironmentType::PUBLIC,
  95         default_api_auth(),
  96     )?;
  97
  98     config.add_alias("novnc", "/usr/share/novnc-pve");
  99     config.add_alias("extjs", "/usr/share/javascript/extjs");
 100     config.add_alias("qrcodejs", "/usr/share/javascript/qrcodejs");
 101     config.add_alias("fontawesome", "/usr/share/fonts-font-awesome");
 102     config.add_alias("xtermjs", "/usr/share/pve-xtermjs");
 103     config.add_alias("locale", "/usr/share/pbs-i18n");
 104     config.add_alias("widgettoolkit", "/usr/share/javascript/proxmox-widget-toolkit");
 105     config.add_alias("docs", "/usr/share/doc/proxmox-backup/html");
 106
 107     let mut indexpath = PathBuf::from(pbs_buildcfg::JS_DIR);
 108     indexpath.push("index.hbs");
 109     config.register_template("index", &indexpath)?;
 110     config.register_template("console", "/usr/share/pve-xtermjs/index.html.hbs")?;
 111
 112     let backup_user = pbs_config::backup_user()?;
 113     let mut commando_sock = proxmox_rest_server::CommandoSocket::new(crate::server::our_ctrl_sock(), backup_user.gid);
 114
 115     let dir_opts = CreateOptions::new().owner(backup_user.uid).group(backup_user.gid);
 116     let file_opts = CreateOptions::new().owner(backup_user.uid).group(backup_user.gid);
 117
 118     config.enable_file_log(
 119         pbs_buildcfg::API_ACCESS_LOG_FN,
 120         Some(dir_opts),
 121         Some(file_opts),
 122         &mut commando_sock,
 123     )?;
 124
 125     let rest_server = RestServer::new(config);
 126
 127     //openssl req -x509 -newkey rsa:4096 -keyout /etc/proxmox-backup/proxy.key -out /etc/proxmox-backup/proxy.pem -nodes
 128
 129     // we build the initial acceptor here as we cannot start if this fails
 130     let acceptor = make_tls_acceptor()?;
 131     let acceptor = Arc::new(Mutex::new(acceptor));
 132
 133     // to renew the acceptor we just add a command-socket handler
 134     commando_sock.register_command(
 135         "reload-certificate".to_string(),
 136         {
 137             let acceptor = Arc::clone(&acceptor);
 138             move |_value| -> Result<_, Error> {
 139                 log::info!("reloading certificate");
 140                 match make_tls_acceptor() {
 141                     Err(err) => log::error!("error reloading certificate: {}", err),
 142                     Ok(new_acceptor) => {
 143                         let mut guard = acceptor.lock().unwrap();
 144                         *guard = new_acceptor;
 145                     }
 146                 }
 147                 Ok(Value::Null)
 148             }
 149         },
 150     )?;
 151
 152     // to remove references for not configured datastores
 153     commando_sock.register_command(
 154         "datastore-removed".to_string(),
 155         |_value| {
 156             if let Err(err) = proxmox_backup::backup::DataStore::remove_unused_datastores() {
 157                 log::error!("could not refresh datastores: {}", err);
 158             }
 159             Ok(Value::Null)
 160         }
 161     )?;
 162
 163     let server = daemon::create_daemon(
 164         ([0,0,0,0,0,0,0,0], 8007).into(),
 165         move |listener, ready| {
 166
 167             let connections = accept_connections(listener, acceptor, debug);
 168             let connections = hyper::server::accept::from_stream(ReceiverStream::new(connections));
 169
 170             Ok(ready
 171                .and_then(|_| hyper::Server::builder(connections)
 172                     .serve(rest_server)
 173                     .with_graceful_shutdown(proxmox_rest_server::shutdown_future())
 174                     .map_err(Error::from)
 175                 )
 176                 .map_err(|err| eprintln!("server error: {}", err))
 177                 .map(|_| ())
 178             )
 179         },
 180         "proxmox-backup-proxy.service",
 181     );
 182
 183     server::write_pid(pbs_buildcfg::PROXMOX_BACKUP_PROXY_PID_FN)?;
 184     daemon::systemd_notify(daemon::SystemdNotify::Ready)?;
 185
 186     let init_result: Result<(), Error> = try_block!({
 187         server::register_task_control_commands(&mut commando_sock)?;
 188         commando_sock.spawn()?;
 189         proxmox_rest_server::server_state_init()?;
 190         Ok(())
 191     });
 192
 193     if let Err(err) = init_result {
 194         bail!("unable to start daemon - {}", err);
 195     }
 196
 197     start_task_scheduler();
 198     start_stat_generator();
 199
 200     server.await?;
 201     log::info!("server shutting down, waiting for active workers to complete");
 202     proxmox_rest_server::last_worker_future().await?;
 203     log::info!("done - exit server");
 204
 205     Ok(())
 206 }
 207
 208 fn make_tls_acceptor() -> Result<SslAcceptor, Error> {
 209     let key_path = configdir!("/proxy.key");
 210     let cert_path = configdir!("/proxy.pem");
 211
 212     let mut acceptor = SslAcceptor::mozilla_intermediate_v5(SslMethod::tls()).unwrap();
 213     acceptor.set_private_key_file(key_path, SslFiletype::PEM)
 214         .map_err(|err| format_err!("unable to read proxy key {} - {}", key_path, err))?;
 215     acceptor.set_certificate_chain_file(cert_path)
 216         .map_err(|err| format_err!("unable to read proxy cert {} - {}", cert_path, err))?;
 217     acceptor.check_private_key().unwrap();
 218
 219     Ok(acceptor.build())
 220 }
 221
 222 type ClientStreamResult =
 223     Result<std::pin::Pin<Box<tokio_openssl::SslStream<tokio::net::TcpStream>>>, Error>;
 224 const MAX_PENDING_ACCEPTS: usize = 1024;
 225
 226 fn accept_connections(
 227     listener: tokio::net::TcpListener,
 228     acceptor: Arc<Mutex<openssl::ssl::SslAcceptor>>,
 229     debug: bool,
 230 ) -> tokio::sync::mpsc::Receiver<ClientStreamResult> {
 231
 232     let (sender, receiver) = tokio::sync::mpsc::channel(MAX_PENDING_ACCEPTS);
 233
 234     tokio::spawn(accept_connection(listener, acceptor, debug, sender));
 235
 236     receiver
 237 }
 238
 239 async fn accept_connection(
 240     listener: tokio::net::TcpListener,
 241     acceptor: Arc<Mutex<openssl::ssl::SslAcceptor>>,
 242     debug: bool,
 243     sender: tokio::sync::mpsc::Sender<ClientStreamResult>,
 244 ) {
 245     let accept_counter = Arc::new(());
 246
 247     loop {
 248         let (sock, _addr) = match listener.accept().await {
 249             Ok(conn) => conn,
 250             Err(err) =>  {
 251                 eprintln!("error accepting tcp connection: {}", err);
 252                 continue;
 253             }
 254         };
 255
 256         sock.set_nodelay(true).unwrap();
 257         let _ = set_tcp_keepalive(sock.as_raw_fd(), PROXMOX_BACKUP_TCP_KEEPALIVE_TIME);
 258
 259         let ssl = { // limit acceptor_guard scope
 260             // Acceptor can be reloaded using the command socket "reload-certificate" command
 261             let acceptor_guard = acceptor.lock().unwrap();
 262
 263             match openssl::ssl::Ssl::new(acceptor_guard.context()) {
 264                 Ok(ssl) => ssl,
 265                 Err(err) => {
 266                     eprintln!("failed to create Ssl object from Acceptor context - {}", err);
 267                     continue;
 268                 },
 269             }
 270         };
 271
 272         let stream = match tokio_openssl::SslStream::new(ssl, sock) {
 273             Ok(stream) => stream,
 274             Err(err) => {
 275                 eprintln!("failed to create SslStream using ssl and connection socket - {}", err);
 276                 continue;
 277             },
 278         };
 279
 280         let mut stream = Box::pin(stream);
 281         let sender = sender.clone();
 282
 283         if Arc::strong_count(&accept_counter) > MAX_PENDING_ACCEPTS {
 284             eprintln!("connection rejected - to many open connections");
 285             continue;
 286         }
 287
 288         let accept_counter = Arc::clone(&accept_counter);
 289         tokio::spawn(async move {
 290             let accept_future = tokio::time::timeout(
 291                 Duration::new(10, 0), stream.as_mut().accept());
 292
 293             let result = accept_future.await;
 294
 295             match result {
 296                 Ok(Ok(())) => {
 297                     if sender.send(Ok(stream)).await.is_err() && debug {
 298                         eprintln!("detect closed connection channel");
 299                     }
 300                 }
 301                 Ok(Err(err)) => {
 302                     if debug {
 303                         eprintln!("https handshake failed - {}", err);
 304                     }
 305                 }
 306                 Err(_) => {
 307                     if debug {
 308                         eprintln!("https handshake timeout");
 309                     }
 310                 }
 311             }
 312
 313             drop(accept_counter); // decrease reference count
 314         });
 315     }
 316 }
 317
 318 fn start_stat_generator() {
 319     let abort_future = proxmox_rest_server::shutdown_future();
 320     let future = Box::pin(run_stat_generator());
 321     let task = futures::future::select(future, abort_future);
 322     tokio::spawn(task.map(|_| ()));
 323 }
 324
 325 fn start_task_scheduler() {
 326     let abort_future = proxmox_rest_server::shutdown_future();
 327     let future = Box::pin(run_task_scheduler());
 328     let task = futures::future::select(future, abort_future);
 329     tokio::spawn(task.map(|_| ()));
 330 }
 331
 332 use std::time::{SystemTime, Instant, Duration, UNIX_EPOCH};
 333
 334 fn next_minute() -> Result<Instant, Error> {
 335     let now = SystemTime::now();
 336     let epoch_now = now.duration_since(UNIX_EPOCH)?;
 337     let epoch_next = Duration::from_secs((epoch_now.as_secs()/60  + 1)*60);
 338     Ok(Instant::now() + epoch_next - epoch_now)
 339 }
 340
 341 async fn run_task_scheduler() {
 342
 343     let mut count: usize = 0;
 344
 345     loop {
 346         count += 1;
 347
 348         let delay_target = match next_minute() {  // try to run very minute
 349             Ok(d) => d,
 350             Err(err) => {
 351                 eprintln!("task scheduler: compute next minute failed - {}", err);
 352                 tokio::time::sleep_until(tokio::time::Instant::from_std(Instant::now() + Duration::from_secs(60))).await;
 353                 continue;
 354             }
 355         };
 356
 357         if count > 2 { // wait 1..2 minutes before starting
 358             match schedule_tasks().catch_unwind().await {
 359                 Err(panic) => {
 360                     match panic.downcast::<&str>() {
 361                         Ok(msg) => {
 362                             eprintln!("task scheduler panic: {}", msg);
 363                         }
 364                         Err(_) => {
 365                             eprintln!("task scheduler panic - unknown type");
 366                         }
 367                     }
 368                 }
 369                 Ok(Err(err)) => {
 370                     eprintln!("task scheduler failed - {:?}", err);
 371                 }
 372                 Ok(Ok(_)) => {}
 373             }
 374         }
 375
 376         tokio::time::sleep_until(tokio::time::Instant::from_std(delay_target)).await;
 377     }
 378 }
 379
 380 async fn schedule_tasks() -> Result<(), Error> {
 381
 382     schedule_datastore_garbage_collection().await;
 383     schedule_datastore_prune().await;
 384     schedule_datastore_sync_jobs().await;
 385     schedule_datastore_verify_jobs().await;
 386     schedule_tape_backup_jobs().await;
 387     schedule_task_log_rotate().await;
 388
 389     Ok(())
 390 }
 391
 392 async fn schedule_datastore_garbage_collection() {
 393
 394     let config = match pbs_config::datastore::config() {
 395         Err(err) => {
 396             eprintln!("unable to read datastore config - {}", err);
 397             return;
 398         }
 399         Ok((config, _digest)) => config,
 400     };
 401
 402     for (store, (_, store_config)) in config.sections {
 403         let datastore = match DataStore::lookup_datastore(&store) {
 404             Ok(datastore) => datastore,
 405             Err(err) => {
 406                 eprintln!("lookup_datastore failed - {}", err);
 407                 continue;
 408             }
 409         };
 410
 411         let store_config: DataStoreConfig = match serde_json::from_value(store_config) {
 412             Ok(c) => c,
 413             Err(err) => {
 414                 eprintln!("datastore config from_value failed - {}", err);
 415                 continue;
 416             }
 417         };
 418
 419         let event_str = match store_config.gc_schedule {
 420             Some(event_str) => event_str,
 421             None => continue,
 422         };
 423
 424         let event = match parse_calendar_event(&event_str) {
 425             Ok(event) => event,
 426             Err(err) => {
 427                 eprintln!("unable to parse schedule '{}' - {}", event_str, err);
 428                 continue;
 429             }
 430         };
 431
 432         if datastore.garbage_collection_running() { continue; }
 433
 434         let worker_type = "garbage_collection";
 435
 436         let last = match jobstate::last_run_time(worker_type, &store) {
 437             Ok(time) => time,
 438             Err(err) => {
 439                 eprintln!("could not get last run time of {} {}: {}", worker_type, store, err);
 440                 continue;
 441             }
 442         };
 443
 444         let next = match compute_next_event(&event, last, false) {
 445             Ok(Some(next)) => next,
 446             Ok(None) => continue,
 447             Err(err) => {
 448                 eprintln!("compute_next_event for '{}' failed - {}", event_str, err);
 449                 continue;
 450             }
 451         };
 452
 453         let now = proxmox::tools::time::epoch_i64();
 454
 455         if next > now  { continue; }
 456
 457         let job = match Job::new(worker_type, &store) {
 458             Ok(job) => job,
 459             Err(_) => continue, // could not get lock
 460         };
 461
 462         let auth_id = Authid::root_auth_id();
 463
 464         if let Err(err) = crate::server::do_garbage_collection_job(job, datastore, auth_id, Some(event_str), false) {
 465             eprintln!("unable to start garbage collection job on datastore {} - {}", store, err);
 466         }
 467     }
 468 }
 469
 470 async fn schedule_datastore_prune() {
 471
 472     let config = match pbs_config::datastore::config() {
 473         Err(err) => {
 474             eprintln!("unable to read datastore config - {}", err);
 475             return;
 476         }
 477         Ok((config, _digest)) => config,
 478     };
 479
 480     for (store, (_, store_config)) in config.sections {
 481
 482         let store_config: DataStoreConfig = match serde_json::from_value(store_config) {
 483             Ok(c) => c,
 484             Err(err) => {
 485                 eprintln!("datastore '{}' config from_value failed - {}", store, err);
 486                 continue;
 487             }
 488         };
 489
 490         let event_str = match store_config.prune_schedule {
 491             Some(event_str) => event_str,
 492             None => continue,
 493         };
 494
 495         let prune_options = PruneOptions {
 496             keep_last: store_config.keep_last,
 497             keep_hourly: store_config.keep_hourly,
 498             keep_daily:  store_config.keep_daily,
 499             keep_weekly: store_config.keep_weekly,
 500             keep_monthly: store_config.keep_monthly,
 501             keep_yearly: store_config.keep_yearly,
 502         };
 503
 504         if !pbs_datastore::prune::keeps_something(&prune_options) { // no prune settings - keep all
 505             continue;
 506         }
 507
 508         let worker_type = "prune";
 509         if check_schedule(worker_type, &event_str, &store) {
 510             let job = match Job::new(worker_type, &store) {
 511                 Ok(job) => job,
 512                 Err(_) => continue, // could not get lock
 513             };
 514
 515             let auth_id = Authid::root_auth_id().clone();
 516             if let Err(err) = do_prune_job(job, prune_options, store.clone(), &auth_id, Some(event_str)) {
 517                 eprintln!("unable to start datastore prune job {} - {}", &store, err);
 518             }
 519         };
 520     }
 521 }
 522
 523 async fn schedule_datastore_sync_jobs() {
 524
 525
 526     let config = match pbs_config::sync::config() {
 527         Err(err) => {
 528             eprintln!("unable to read sync job config - {}", err);
 529             return;
 530         }
 531         Ok((config, _digest)) => config,
 532     };
 533
 534     for (job_id, (_, job_config)) in config.sections {
 535         let job_config: SyncJobConfig = match serde_json::from_value(job_config) {
 536             Ok(c) => c,
 537             Err(err) => {
 538                 eprintln!("sync job config from_value failed - {}", err);
 539                 continue;
 540             }
 541         };
 542
 543         let event_str = match job_config.schedule {
 544             Some(ref event_str) => event_str.clone(),
 545             None => continue,
 546         };
 547
 548         let worker_type = "syncjob";
 549         if check_schedule(worker_type, &event_str, &job_id) {
 550             let job = match Job::new(worker_type, &job_id) {
 551                 Ok(job) => job,
 552                 Err(_) => continue, // could not get lock
 553             };
 554
 555             let auth_id = Authid::root_auth_id().clone();
 556             if let Err(err) = do_sync_job(job, job_config, &auth_id, Some(event_str)) {
 557                 eprintln!("unable to start datastore sync job {} - {}", &job_id, err);
 558             }
 559         };
 560     }
 561 }
 562
 563 async fn schedule_datastore_verify_jobs() {
 564
 565     let config = match pbs_config::verify::config() {
 566         Err(err) => {
 567             eprintln!("unable to read verification job config - {}", err);
 568             return;
 569         }
 570         Ok((config, _digest)) => config,
 571     };
 572     for (job_id, (_, job_config)) in config.sections {
 573         let job_config: VerificationJobConfig = match serde_json::from_value(job_config) {
 574             Ok(c) => c,
 575             Err(err) => {
 576                 eprintln!("verification job config from_value failed - {}", err);
 577                 continue;
 578             }
 579         };
 580         let event_str = match job_config.schedule {
 581             Some(ref event_str) => event_str.clone(),
 582             None => continue,
 583         };
 584
 585         let worker_type = "verificationjob";
 586         let auth_id = Authid::root_auth_id().clone();
 587         if check_schedule(worker_type, &event_str, &job_id) {
 588             let job = match Job::new(&worker_type, &job_id) {
 589                 Ok(job) => job,
 590                 Err(_) => continue, // could not get lock
 591             };
 592             if let Err(err) = do_verification_job(job, job_config, &auth_id, Some(event_str)) {
 593                 eprintln!("unable to start datastore verification job {} - {}", &job_id, err);
 594             }
 595         };
 596     }
 597 }
 598
 599 async fn schedule_tape_backup_jobs() {
 600
 601     let config = match pbs_config::tape_job::config() {
 602         Err(err) => {
 603             eprintln!("unable to read tape job config - {}", err);
 604             return;
 605         }
 606         Ok((config, _digest)) => config,
 607     };
 608     for (job_id, (_, job_config)) in config.sections {
 609         let job_config: TapeBackupJobConfig = match serde_json::from_value(job_config) {
 610             Ok(c) => c,
 611             Err(err) => {
 612                 eprintln!("tape backup job config from_value failed - {}", err);
 613                 continue;
 614             }
 615         };
 616         let event_str = match job_config.schedule {
 617             Some(ref event_str) => event_str.clone(),
 618             None => continue,
 619         };
 620
 621         let worker_type = "tape-backup-job";
 622         let auth_id = Authid::root_auth_id().clone();
 623         if check_schedule(worker_type, &event_str, &job_id) {
 624             let job = match Job::new(&worker_type, &job_id) {
 625                 Ok(job) => job,
 626                 Err(_) => continue, // could not get lock
 627             };
 628             if let Err(err) = do_tape_backup_job(job, job_config.setup, &auth_id, Some(event_str)) {
 629                 eprintln!("unable to start tape backup job {} - {}", &job_id, err);
 630             }
 631         };
 632     }
 633 }
 634
 635
 636 async fn schedule_task_log_rotate() {
 637
 638     let worker_type = "logrotate";
 639     let job_id = "access-log_and_task-archive";
 640
 641     // schedule daily at 00:00 like normal logrotate
 642     let schedule = "00:00";
 643
 644     if !check_schedule(worker_type, schedule, job_id) {
 645         // if we never ran the rotation, schedule instantly
 646         match jobstate::JobState::load(worker_type, job_id) {
 647             Ok(state) => match state {
 648                 jobstate::JobState::Created { .. } => {},
 649                 _ => return,
 650             },
 651             _ => return,
 652         }
 653     }
 654
 655     let mut job = match Job::new(worker_type, job_id) {
 656         Ok(job) => job,
 657         Err(_) => return, // could not get lock
 658     };
 659
 660     if let Err(err) = WorkerTask::new_thread(
 661         worker_type,
 662         None,
 663         Authid::root_auth_id().clone(),
 664         false,
 665         move |worker| {
 666             job.start(&worker.upid().to_string())?;
 667             worker.log("starting task log rotation".to_string());
 668
 669             let result = try_block!({
 670                 let max_size = 512 * 1024 - 1; // an entry has ~ 100b, so > 5000 entries/file
 671                 let max_files = 20; // times twenty files gives > 100000 task entries
 672                 let has_rotated = rotate_task_log_archive(max_size, true, Some(max_files))?;
 673                 if has_rotated {
 674                     worker.log("task log archive was rotated".to_string());
 675                 } else {
 676                     worker.log("task log archive was not rotated".to_string());
 677                 }
 678
 679                 let max_size = 32 * 1024 * 1024 - 1;
 680                 let max_files = 14;
 681                 let mut logrotate = LogRotate::new(pbs_buildcfg::API_ACCESS_LOG_FN, true)
 682                         .ok_or_else(|| format_err!("could not get API access log file names"))?;
 683
 684                 if logrotate.rotate(max_size, None, Some(max_files))? {
 685                     println!("rotated access log, telling daemons to re-open log file");
 686                     pbs_runtime::block_on(command_reopen_logfiles())?;
 687                     worker.log("API access log was rotated".to_string());
 688                 } else {
 689                     worker.log("API access log was not rotated".to_string());
 690                 }
 691
 692                 let mut logrotate = LogRotate::new(pbs_buildcfg::API_AUTH_LOG_FN, true)
 693                         .ok_or_else(|| format_err!("could not get API auth log file names"))?;
 694
 695                 if logrotate.rotate(max_size, None, Some(max_files))? {
 696                     worker.log("API authentication log was rotated".to_string());
 697                 } else {
 698                     worker.log("API authentication log was not rotated".to_string());
 699                 }
 700
 701                 Ok(())
 702             });
 703
 704             let status = worker.create_state(&result);
 705
 706             if let Err(err) = job.finish(status) {
 707                 eprintln!("could not finish job state for {}: {}", worker_type, err);
 708             }
 709
 710             result
 711         },
 712     ) {
 713         eprintln!("unable to start task log rotation: {}", err);
 714     }
 715
 716 }
 717
 718 async fn command_reopen_logfiles() -> Result<(), Error> {
 719     // only care about the most recent daemon instance for each, proxy & api, as other older ones
 720     // should not respond to new requests anyway, but only finish their current one and then exit.
 721     let sock = crate::server::our_ctrl_sock();
 722     let f1 = proxmox_rest_server::send_command(sock, "{\"command\":\"api-access-log-reopen\"}\n");
 723
 724     let pid = crate::server::read_pid(pbs_buildcfg::PROXMOX_BACKUP_API_PID_FN)?;
 725     let sock = crate::server::ctrl_sock_from_pid(pid);
 726     let f2 = proxmox_rest_server::send_command(sock, "{\"command\":\"api-access-log-reopen\"}\n");
 727
 728     match futures::join!(f1, f2) {
 729         (Err(e1), Err(e2)) => Err(format_err!("reopen commands failed, proxy: {}; api: {}", e1, e2)),
 730         (Err(e1), Ok(_)) => Err(format_err!("reopen commands failed, proxy: {}", e1)),
 731         (Ok(_), Err(e2)) => Err(format_err!("reopen commands failed, api: {}", e2)),
 732         _ => Ok(()),
 733     }
 734 }
 735
 736 async fn run_stat_generator() {
 737
 738     let mut count = 0;
 739     loop {
 740         count += 1;
 741         let save = if count >= 6 { count = 0; true } else { false };
 742
 743         let delay_target = Instant::now() +  Duration::from_secs(10);
 744
 745         generate_host_stats(save).await;
 746
 747         tokio::time::sleep_until(tokio::time::Instant::from_std(delay_target)).await;
 748
 749      }
 750
 751 }
 752
 753 fn rrd_update_gauge(name: &str, value: f64, save: bool) {
 754     use proxmox_backup::rrd;
 755     if let Err(err) = rrd::update_value(name, value, rrd::DST::Gauge, save) {
 756         eprintln!("rrd::update_value '{}' failed - {}", name, err);
 757     }
 758 }
 759
 760 fn rrd_update_derive(name: &str, value: f64, save: bool) {
 761     use proxmox_backup::rrd;
 762     if let Err(err) = rrd::update_value(name, value, rrd::DST::Derive, save) {
 763         eprintln!("rrd::update_value '{}' failed - {}", name, err);
 764     }
 765 }
 766
 767 async fn generate_host_stats(save: bool) {
 768     use proxmox::sys::linux::procfs::{
 769         read_meminfo, read_proc_stat, read_proc_net_dev, read_loadavg};
 770
 771     pbs_runtime::block_in_place(move || {
 772
 773         match read_proc_stat() {
 774             Ok(stat) => {
 775                 rrd_update_gauge("host/cpu", stat.cpu, save);
 776                 rrd_update_gauge("host/iowait", stat.iowait_percent, save);
 777             }
 778             Err(err) => {
 779                 eprintln!("read_proc_stat failed - {}", err);
 780             }
 781         }
 782
 783         match read_meminfo() {
 784             Ok(meminfo) => {
 785                 rrd_update_gauge("host/memtotal", meminfo.memtotal as f64, save);
 786                 rrd_update_gauge("host/memused", meminfo.memused as f64, save);
 787                 rrd_update_gauge("host/swaptotal", meminfo.swaptotal as f64, save);
 788                 rrd_update_gauge("host/swapused", meminfo.swapused as f64, save);
 789             }
 790             Err(err) => {
 791                 eprintln!("read_meminfo failed - {}", err);
 792             }
 793         }
 794
 795         match read_proc_net_dev() {
 796             Ok(netdev) => {
 797                 use pbs_config::network::is_physical_nic;
 798                 let mut netin = 0;
 799                 let mut netout = 0;
 800                 for item in netdev {
 801                     if !is_physical_nic(&item.device) { continue; }
 802                     netin += item.receive;
 803                     netout += item.send;
 804                 }
 805                 rrd_update_derive("host/netin", netin as f64, save);
 806                 rrd_update_derive("host/netout", netout as f64, save);
 807             }
 808             Err(err) => {
 809                 eprintln!("read_prox_net_dev failed - {}", err);
 810             }
 811         }
 812
 813         match read_loadavg() {
 814             Ok(loadavg) => {
 815                 rrd_update_gauge("host/loadavg", loadavg.0 as f64, save);
 816             }
 817             Err(err) => {
 818                 eprintln!("read_loadavg failed - {}", err);
 819             }
 820         }
 821
 822         let disk_manager = DiskManage::new();
 823
 824         gather_disk_stats(disk_manager.clone(), Path::new("/"), "host", save);
 825
 826         match pbs_config::datastore::config() {
 827             Ok((config, _)) => {
 828                 let datastore_list: Vec<DataStoreConfig> =
 829                     config.convert_to_typed_array("datastore").unwrap_or_default();
 830
 831                 for config in datastore_list {
 832
 833                     let rrd_prefix = format!("datastore/{}", config.name);
 834                     let path = std::path::Path::new(&config.path);
 835                     gather_disk_stats(disk_manager.clone(), path, &rrd_prefix, save);
 836                 }
 837             }
 838             Err(err) => {
 839                 eprintln!("read datastore config failed - {}", err);
 840             }
 841         }
 842
 843     });
 844 }
 845
 846 fn check_schedule(worker_type: &str, event_str: &str, id: &str) -> bool {
 847     let event = match parse_calendar_event(event_str) {
 848         Ok(event) => event,
 849         Err(err) => {
 850             eprintln!("unable to parse schedule '{}' - {}", event_str, err);
 851             return false;
 852         }
 853     };
 854
 855     let last = match jobstate::last_run_time(worker_type, &id) {
 856         Ok(time) => time,
 857         Err(err) => {
 858             eprintln!("could not get last run time of {} {}: {}", worker_type, id, err);
 859             return false;
 860         }
 861     };
 862
 863     let next = match compute_next_event(&event, last, false) {
 864         Ok(Some(next)) => next,
 865         Ok(None) => return false,
 866         Err(err) => {
 867             eprintln!("compute_next_event for '{}' failed - {}", event_str, err);
 868             return false;
 869         }
 870     };
 871
 872     let now = proxmox::tools::time::epoch_i64();
 873     next <= now
 874 }
 875
 876 fn gather_disk_stats(disk_manager: Arc<DiskManage>, path: &Path, rrd_prefix: &str, save: bool) {
 877
 878     match proxmox_backup::tools::disks::disk_usage(path) {
 879         Ok(status) => {
 880             let rrd_key = format!("{}/total", rrd_prefix);
 881             rrd_update_gauge(&rrd_key, status.total as f64, save);
 882             let rrd_key = format!("{}/used", rrd_prefix);
 883             rrd_update_gauge(&rrd_key, status.used as f64, save);
 884         }
 885         Err(err) => {
 886             eprintln!("read disk_usage on {:?} failed - {}", path, err);
 887         }
 888     }
 889
 890     match disk_manager.find_mounted_device(path) {
 891         Ok(None) => {},
 892         Ok(Some((fs_type, device, source))) => {
 893             let mut device_stat = None;
 894             match fs_type.as_str() {
 895                 "zfs" => {
 896                     if let Some(source) = source {
 897                         let pool = get_pool_from_dataset(&source).unwrap_or(&source);
 898                         match zfs_pool_stats(pool) {
 899                             Ok(stat) => device_stat = stat,
 900                             Err(err) => eprintln!("zfs_pool_stats({:?}) failed - {}", pool, err),
 901                         }
 902                     }
 903                 }
 904                 _ => {
 905                     if let Ok(disk) = disk_manager.clone().disk_by_dev_num(device.into_dev_t()) {
 906                         match disk.read_stat() {
 907                             Ok(stat) => device_stat = stat,
 908                             Err(err) => eprintln!("disk.read_stat {:?} failed - {}", path, err),
 909                         }
 910                     }
 911                 }
 912             }
 913             if let Some(stat) = device_stat {
 914                 let rrd_key = format!("{}/read_ios", rrd_prefix);
 915                 rrd_update_derive(&rrd_key, stat.read_ios as f64, save);
 916                 let rrd_key = format!("{}/read_bytes", rrd_prefix);
 917                 rrd_update_derive(&rrd_key, (stat.read_sectors*512) as f64, save);
 918
 919                 let rrd_key = format!("{}/write_ios", rrd_prefix);
 920                 rrd_update_derive(&rrd_key, stat.write_ios as f64, save);
 921                 let rrd_key = format!("{}/write_bytes", rrd_prefix);
 922                 rrd_update_derive(&rrd_key, (stat.write_sectors*512) as f64, save);
 923
 924                 let rrd_key = format!("{}/io_ticks", rrd_prefix);
 925                 rrd_update_derive(&rrd_key, (stat.io_ticks as f64)/1000.0, save);
 926             }
 927         }
 928         Err(err) => {
 929             eprintln!("find_mounted_device failed - {}", err);
 930         }
 931     }
 932 }