src/bin/proxmox-backup-proxy.rs

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