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