2 use std
::ffi
::OsString
;
5 use anyhow
::{bail, format_err, Error}
;
8 use openssl
::ssl
::{SslMethod, SslAcceptor, SslFiletype}
;
10 use proxmox
::try_block
;
11 use proxmox
::api
::RpcEnvironmentType
;
12 use proxmox
::sys
::linux
::procfs
::mountinfo
::{Device, MountInfo}
;
14 use proxmox_backup
::configdir
;
15 use proxmox_backup
::buildcfg
;
16 use proxmox_backup
::server
;
17 use proxmox_backup
::tools
::daemon
;
18 use proxmox_backup
::server
::{ApiConfig, rest::*}
;
19 use proxmox_backup
::auth_helpers
::*;
20 use proxmox_backup
::tools
::disks
::{ DiskManage, zfs::zfs_pool_stats }
;
23 if let Err(err
) = proxmox_backup
::tools
::runtime
::main(run()) {
24 eprintln
!("Error: {}", err
);
25 std
::process
::exit(-1);
29 async
fn run() -> Result
<(), Error
> {
30 if let Err(err
) = syslog
::init(
31 syslog
::Facility
::LOG_DAEMON
,
32 log
::LevelFilter
::Info
,
33 Some("proxmox-backup-proxy")) {
34 bail
!("unable to inititialize syslog - {}", err
);
37 let _
= public_auth_key(); // load with lazy_static
38 let _
= csrf_secret(); // load with lazy_static
40 let mut config
= ApiConfig
::new(
41 buildcfg
::JS_DIR
, &proxmox_backup
::api2
::ROUTER
, RpcEnvironmentType
::PUBLIC
)?
;
43 // add default dirs which includes jquery and bootstrap
44 // my $base = '/usr/share/libpve-http-server-perl';
45 // add_dirs($self->{dirs}, '/css/' => "$base/css/");
46 // add_dirs($self->{dirs}, '/js/' => "$base/js/");
47 // add_dirs($self->{dirs}, '/fonts/' => "$base/fonts/");
48 config
.add_alias("novnc", "/usr/share/novnc-pve");
49 config
.add_alias("extjs", "/usr/share/javascript/extjs");
50 config
.add_alias("fontawesome", "/usr/share/fonts-font-awesome");
51 config
.add_alias("xtermjs", "/usr/share/pve-xtermjs");
52 config
.add_alias("widgettoolkit", "/usr/share/javascript/proxmox-widget-toolkit");
53 config
.add_alias("css", "/usr/share/javascript/proxmox-backup/css");
54 config
.add_alias("docs", "/usr/share/doc/proxmox-backup/html");
56 let rest_server
= RestServer
::new(config
);
58 //openssl req -x509 -newkey rsa:4096 -keyout /etc/proxmox-backup/proxy.key -out /etc/proxmox-backup/proxy.pem -nodes
59 let key_path
= configdir
!("/proxy.key");
60 let cert_path
= configdir
!("/proxy.pem");
62 let mut acceptor
= SslAcceptor
::mozilla_intermediate(SslMethod
::tls()).unwrap();
63 acceptor
.set_private_key_file(key_path
, SslFiletype
::PEM
)
64 .map_err(|err
| format_err
!("unable to read proxy key {} - {}", key_path
, err
))?
;
65 acceptor
.set_certificate_chain_file(cert_path
)
66 .map_err(|err
| format_err
!("unable to read proxy cert {} - {}", cert_path
, err
))?
;
67 acceptor
.check_private_key().unwrap();
69 let acceptor
= Arc
::new(acceptor
.build());
71 let server
= daemon
::create_daemon(
72 ([0,0,0,0,0,0,0,0], 8007).into(),
74 let connections
= proxmox_backup
::tools
::async_io
::StaticIncoming
::from(listener
)
76 .try_filter_map(move |(sock
, _addr
)| {
77 let acceptor
= Arc
::clone(&acceptor
);
79 sock
.set_nodelay(true).unwrap();
80 sock
.set_send_buffer_size(1024*1024).unwrap();
81 sock
.set_recv_buffer_size(1024*1024).unwrap();
82 Ok(tokio_openssl
::accept(&acceptor
, sock
)
84 .ok() // handshake errors aren't be fatal, so return None to filter
88 let connections
= proxmox_backup
::tools
::async_io
::HyperAccept(connections
);
91 .and_then(|_
| hyper
::Server
::builder(connections
)
93 .with_graceful_shutdown(server
::shutdown_future())
96 .map_err(|err
| eprintln
!("server error: {}", err
))
102 daemon
::systemd_notify(daemon
::SystemdNotify
::Ready
)?
;
104 let init_result
: Result
<(), Error
> = try_block
!({
105 server
::create_task_control_socket()?
;
106 server
::server_state_init()?
;
110 if let Err(err
) = init_result
{
111 bail
!("unable to start daemon - {}", err
);
114 start_task_scheduler();
115 start_stat_generator();
118 log
::info
!("server shutting down, waiting for active workers to complete");
119 proxmox_backup
::server
::last_worker_future().await?
;
120 log
::info
!("done - exit server");
125 fn start_stat_generator() {
126 let abort_future
= server
::shutdown_future();
127 let future
= Box
::pin(run_stat_generator());
128 let task
= futures
::future
::select(future
, abort_future
);
129 tokio
::spawn(task
.map(|_
| ()));
132 fn start_task_scheduler() {
133 let abort_future
= server
::shutdown_future();
134 let future
= Box
::pin(run_task_scheduler());
135 let task
= futures
::future
::select(future
, abort_future
);
136 tokio
::spawn(task
.map(|_
| ()));
139 use std
::time
:: {Instant, Duration, SystemTime, UNIX_EPOCH}
;
141 fn next_minute() -> Result
<Instant
, Error
> {
142 let epoch_now
= SystemTime
::now().duration_since(UNIX_EPOCH
)?
;
143 let epoch_next
= Duration
::from_secs((epoch_now
.as_secs()/60 + 1)*60);
144 Ok(Instant
::now() + epoch_next
- epoch_now
)
147 async
fn run_task_scheduler() {
149 let mut count
: usize = 0;
154 let delay_target
= match next_minute() { // try to run very minute
157 eprintln
!("task scheduler: compute next minute failed - {}", err
);
158 tokio
::time
::delay_until(tokio
::time
::Instant
::from_std(Instant
::now() + Duration
::from_secs(60))).await
;
163 if count
> 2 { // wait 1..2 minutes before starting
164 match schedule_tasks().catch_unwind().await
{
166 match panic
.downcast
::<&str>() {
168 eprintln
!("task scheduler panic: {}", msg
);
171 eprintln
!("task scheduler panic - unknown type");
176 eprintln
!("task scheduler failed - {:?}", err
);
182 tokio
::time
::delay_until(tokio
::time
::Instant
::from_std(delay_target
)).await
;
186 async
fn schedule_tasks() -> Result
<(), Error
> {
188 schedule_datastore_garbage_collection().await
;
189 schedule_datastore_prune().await
;
190 schedule_datastore_sync_jobs().await
;
195 fn lookup_last_worker(worker_type
: &str, worker_id
: &str) -> Result
<Option
<server
::UPID
>, Error
> {
197 let list
= proxmox_backup
::server
::read_task_list()?
;
199 let mut last
: Option
<&server
::UPID
> = None
;
201 for entry
in list
.iter() {
202 if entry
.upid
.worker_type
== worker_type
{
203 if let Some(ref id
) = entry
.upid
.worker_id
{
207 if upid
.starttime
< entry
.upid
.starttime
{
208 last
= Some(&entry
.upid
)
212 last
= Some(&entry
.upid
)
224 async
fn schedule_datastore_garbage_collection() {
226 use proxmox_backup
::backup
::DataStore
;
227 use proxmox_backup
::server
::{UPID, WorkerTask}
;
228 use proxmox_backup
::config
::datastore
::{self, DataStoreConfig}
;
229 use proxmox_backup
::tools
::systemd
::time
::{
230 parse_calendar_event
, compute_next_event
};
232 let config
= match datastore
::config() {
234 eprintln
!("unable to read datastore config - {}", err
);
237 Ok((config
, _digest
)) => config
,
240 for (store
, (_
, store_config
)) in config
.sections
{
241 let datastore
= match DataStore
::lookup_datastore(&store
) {
242 Ok(datastore
) => datastore
,
244 eprintln
!("lookup_datastore failed - {}", err
);
249 let store_config
: DataStoreConfig
= match serde_json
::from_value(store_config
) {
252 eprintln
!("datastore config from_value failed - {}", err
);
257 let event_str
= match store_config
.gc_schedule
{
258 Some(event_str
) => event_str
,
262 let event
= match parse_calendar_event(&event_str
) {
265 eprintln
!("unable to parse schedule '{}' - {}", event_str
, err
);
270 if datastore
.garbage_collection_running() { continue; }
272 let worker_type
= "garbage_collection";
274 let stat
= datastore
.last_gc_status();
275 let last
= if let Some(upid_str
) = stat
.upid
{
276 match upid_str
.parse
::<UPID
>() {
277 Ok(upid
) => upid
.starttime
,
279 eprintln
!("unable to parse upid '{}' - {}", upid_str
, err
);
284 match lookup_last_worker(worker_type
, &store
) {
285 Ok(Some(upid
)) => upid
.starttime
,
288 eprintln
!("lookup_last_job_start failed: {}", err
);
294 let next
= match compute_next_event(&event
, last
, false) {
297 eprintln
!("compute_next_event for '{}' failed - {}", event_str
, err
);
301 let now
= match SystemTime
::now().duration_since(UNIX_EPOCH
) {
302 Ok(epoch_now
) => epoch_now
.as_secs() as i64,
304 eprintln
!("query system time failed - {}", err
);
308 if next
> now { continue; }
310 let store2
= store
.clone();
312 if let Err(err
) = WorkerTask
::new_thread(
318 worker
.log(format
!("starting garbage collection on store {}", store
));
319 worker
.log(format
!("task triggered by schedule '{}'", event_str
));
320 datastore
.garbage_collection(&worker
)
323 eprintln
!("unable to start garbage collection on store {} - {}", store2
, err
);
328 async
fn schedule_datastore_prune() {
330 use proxmox_backup
::backup
::{
331 PruneOptions
, DataStore
, BackupGroup
, BackupDir
, compute_prune_info
};
332 use proxmox_backup
::server
::{WorkerTask}
;
333 use proxmox_backup
::config
::datastore
::{self, DataStoreConfig}
;
334 use proxmox_backup
::tools
::systemd
::time
::{
335 parse_calendar_event
, compute_next_event
};
337 let config
= match datastore
::config() {
339 eprintln
!("unable to read datastore config - {}", err
);
342 Ok((config
, _digest
)) => config
,
345 for (store
, (_
, store_config
)) in config
.sections
{
346 let datastore
= match DataStore
::lookup_datastore(&store
) {
347 Ok(datastore
) => datastore
,
349 eprintln
!("lookup_datastore '{}' failed - {}", store
, err
);
354 let store_config
: DataStoreConfig
= match serde_json
::from_value(store_config
) {
357 eprintln
!("datastore '{}' config from_value failed - {}", store
, err
);
362 let event_str
= match store_config
.prune_schedule
{
363 Some(event_str
) => event_str
,
367 let prune_options
= PruneOptions
{
368 keep_last
: store_config
.keep_last
,
369 keep_hourly
: store_config
.keep_hourly
,
370 keep_daily
: store_config
.keep_daily
,
371 keep_weekly
: store_config
.keep_weekly
,
372 keep_monthly
: store_config
.keep_monthly
,
373 keep_yearly
: store_config
.keep_yearly
,
376 if !prune_options
.keeps_something() { // no prune settings - keep all
380 let event
= match parse_calendar_event(&event_str
) {
383 eprintln
!("unable to parse schedule '{}' - {}", event_str
, err
);
388 //fixme: if last_prune_job_stzill_running { continue; }
390 let worker_type
= "prune";
392 let last
= match lookup_last_worker(worker_type
, &store
) {
393 Ok(Some(upid
)) => upid
.starttime
,
396 eprintln
!("lookup_last_job_start failed: {}", err
);
401 let next
= match compute_next_event(&event
, last
, false) {
404 eprintln
!("compute_next_event for '{}' failed - {}", event_str
, err
);
409 let now
= match SystemTime
::now().duration_since(UNIX_EPOCH
) {
410 Ok(epoch_now
) => epoch_now
.as_secs() as i64,
412 eprintln
!("query system time failed - {}", err
);
416 if next
> now { continue; }
418 let store2
= store
.clone();
420 if let Err(err
) = WorkerTask
::new_thread(
426 worker
.log(format
!("Starting datastore prune on store \"{}\"", store
));
427 worker
.log(format
!("task triggered by schedule '{}'", event_str
));
428 worker
.log(format
!("retention options: {}", prune_options
.cli_options_string()));
430 let base_path
= datastore
.base_path();
432 let groups
= BackupGroup
::list_groups(&base_path
)?
;
433 for group
in groups
{
434 let list
= group
.list_backups(&base_path
)?
;
435 let mut prune_info
= compute_prune_info(list
, &prune_options
)?
;
436 prune_info
.reverse(); // delete older snapshots first
438 worker
.log(format
!("Starting prune on store \"{}\" group \"{}/{}\"",
439 store
, group
.backup_type(), group
.backup_id()));
441 for (info
, keep
) in prune_info
{
444 if keep { "keep" }
else { "remove" }
,
445 group
.backup_type(), group
.backup_id(),
446 BackupDir
::backup_time_to_string(info
.backup_dir
.backup_time())));
449 datastore
.remove_backup_dir(&info
.backup_dir
)?
;
457 eprintln
!("unable to start datastore prune on store {} - {}", store2
, err
);
462 async
fn schedule_datastore_sync_jobs() {
464 use proxmox_backup
::{
466 client
::{ HttpClient, HttpClientOptions, BackupRepository, pull::pull_store }
,
467 server
::{ WorkerTask }
,
468 config
::{ sync::{self, SyncJobConfig}
, remote
::{self, Remote}
},
469 tools
::systemd
::time
::{ parse_calendar_event, compute_next_event }
,
472 let config
= match sync
::config() {
474 eprintln
!("unable to read sync job config - {}", err
);
477 Ok((config
, _digest
)) => config
,
480 let remote_config
= match remote
::config() {
482 eprintln
!("unable to read remote config - {}", err
);
485 Ok((config
, _digest
)) => config
,
488 for (job_id
, (_
, job_config
)) in config
.sections
{
489 let job_config
: SyncJobConfig
= match serde_json
::from_value(job_config
) {
492 eprintln
!("sync job config from_value failed - {}", err
);
497 let event_str
= match job_config
.schedule
{
498 Some(ref event_str
) => event_str
.clone(),
502 let event
= match parse_calendar_event(&event_str
) {
505 eprintln
!("unable to parse schedule '{}' - {}", event_str
, err
);
510 //fixme: if last_sync_job_still_running { continue; }
512 let worker_type
= "sync";
514 let last
= match lookup_last_worker(worker_type
, &job_id
) {
515 Ok(Some(upid
)) => upid
.starttime
,
518 eprintln
!("lookup_last_job_start failed: {}", err
);
523 let next
= match compute_next_event(&event
, last
, false) {
526 eprintln
!("compute_next_event for '{}' failed - {}", event_str
, err
);
531 let now
= match SystemTime
::now().duration_since(UNIX_EPOCH
) {
532 Ok(epoch_now
) => epoch_now
.as_secs() as i64,
534 eprintln
!("query system time failed - {}", err
);
538 if next
> now { continue; }
541 let job_id2
= job_id
.clone();
543 let tgt_store
= match DataStore
::lookup_datastore(&job_config
.store
) {
544 Ok(datastore
) => datastore
,
546 eprintln
!("lookup_datastore '{}' failed - {}", job_config
.store
, err
);
551 let remote
: Remote
= match remote_config
.lookup("remote", &job_config
.remote
) {
552 Ok(remote
) => remote
,
554 eprintln
!("remote_config lookup failed: {}", err
);
559 let username
= String
::from("backup@pam");
561 let delete
= job_config
.remove_vanished
.unwrap_or(true);
563 if let Err(err
) = WorkerTask
::spawn(
565 Some(job_id
.clone()),
568 move |worker
| async
move {
569 worker
.log(format
!("Starting datastore sync job '{}'", job_id
));
570 worker
.log(format
!("task triggered by schedule '{}'", event_str
));
571 worker
.log(format
!("Sync datastore '{}' from '{}/{}'",
572 job_config
.store
, job_config
.remote
, job_config
.remote_store
));
574 let options
= HttpClientOptions
::new()
575 .password(Some(remote
.password
.clone()))
576 .fingerprint(remote
.fingerprint
.clone());
578 let client
= HttpClient
::new(&remote
.host
, &remote
.userid
, options
)?
;
579 let _auth_info
= client
.login() // make sure we can auth
581 .map_err(|err
| format_err
!("remote connection to '{}' failed - {}", remote
.host
, err
))?
;
583 let src_repo
= BackupRepository
::new(Some(remote
.userid
), Some(remote
.host
), job_config
.remote_store
);
585 pull_store(&worker
, &client
, &src_repo
, tgt_store
, delete
, username
).await?
;
590 eprintln
!("unable to start datastore sync job {} - {}", job_id2
, err
);
595 async
fn run_stat_generator() {
598 let delay_target
= Instant
::now() + Duration
::from_secs(10);
600 generate_host_stats().await
;
602 tokio
::time
::delay_until(tokio
::time
::Instant
::from_std(delay_target
)).await
;
607 fn rrd_update_gauge(name
: &str, value
: f64) {
608 use proxmox_backup
::rrd
;
609 if let Err(err
) = rrd
::update_value(name
, value
, rrd
::DST
::Gauge
) {
610 eprintln
!("rrd::update_value '{}' failed - {}", name
, err
);
614 fn rrd_update_derive(name
: &str, value
: f64) {
615 use proxmox_backup
::rrd
;
616 if let Err(err
) = rrd
::update_value(name
, value
, rrd
::DST
::Derive
) {
617 eprintln
!("rrd::update_value '{}' failed - {}", name
, err
);
621 async
fn generate_host_stats() {
622 use proxmox
::sys
::linux
::procfs
::{
623 read_meminfo
, read_proc_stat
, read_proc_net_dev
, read_loadavg
};
624 use proxmox_backup
::config
::datastore
;
627 proxmox_backup
::tools
::runtime
::block_in_place(move || {
629 match read_proc_stat() {
631 rrd_update_gauge("host/cpu", stat
.cpu
);
632 rrd_update_gauge("host/iowait", stat
.iowait_percent
);
635 eprintln
!("read_proc_stat failed - {}", err
);
639 match read_meminfo() {
641 rrd_update_gauge("host/memtotal", meminfo
.memtotal
as f64);
642 rrd_update_gauge("host/memused", meminfo
.memused
as f64);
643 rrd_update_gauge("host/swaptotal", meminfo
.swaptotal
as f64);
644 rrd_update_gauge("host/swapused", meminfo
.swapused
as f64);
647 eprintln
!("read_meminfo failed - {}", err
);
651 match read_proc_net_dev() {
653 use proxmox_backup
::config
::network
::is_physical_nic
;
657 if !is_physical_nic(&item
.device
) { continue; }
658 netin
+= item
.receive
;
661 rrd_update_derive("host/netin", netin
as f64);
662 rrd_update_derive("host/netout", netout
as f64);
665 eprintln
!("read_prox_net_dev failed - {}", err
);
669 match read_loadavg() {
671 rrd_update_gauge("host/loadavg", loadavg
.0 as f64);
674 eprintln
!("read_loadavg failed - {}", err
);
678 let disk_manager
= DiskManage
::new();
680 gather_disk_stats(disk_manager
.clone(), Path
::new("/"), "host");
682 match datastore
::config() {
684 let datastore_list
: Vec
<datastore
::DataStoreConfig
> =
685 config
.convert_to_typed_array("datastore").unwrap_or(Vec
::new());
687 for config
in datastore_list
{
689 let rrd_prefix
= format
!("datastore/{}", config
.name
);
690 let path
= std
::path
::Path
::new(&config
.path
);
691 gather_disk_stats(disk_manager
.clone(), path
, &rrd_prefix
);
695 eprintln
!("read datastore config failed - {}", err
);
703 fn gather_disk_stats(disk_manager
: Arc
<DiskManage
>, path
: &Path
, rrd_prefix
: &str) {
705 match disk_usage(path
) {
706 Ok((total
, used
, _avail
)) => {
707 let rrd_key
= format
!("{}/total", rrd_prefix
);
708 rrd_update_gauge(&rrd_key
, total
as f64);
709 let rrd_key
= format
!("{}/used", rrd_prefix
);
710 rrd_update_gauge(&rrd_key
, used
as f64);
713 eprintln
!("read disk_usage on {:?} failed - {}", path
, err
);
717 match disk_manager
.mount_info() {
719 if let Some((fs_type
, device
, source
)) = find_mounted_device(mountinfo
, path
) {
720 let mut device_stat
= None
;
721 match fs_type
.as_str() {
723 if let Some(pool
) = source
{
724 match zfs_pool_stats(&pool
) {
725 Ok(stat
) => device_stat
= stat
,
726 Err(err
) => eprintln
!("zfs_pool_stats({:?}) failed - {}", pool
, err
),
731 if let Ok(disk
) = disk_manager
.clone().disk_by_dev_num(device
.into_dev_t()) {
732 match disk
.read_stat() {
733 Ok(stat
) => device_stat
= stat
,
734 Err(err
) => eprintln
!("disk.read_stat {:?} failed - {}", path
, err
),
739 if let Some(stat
) = device_stat
{
740 let rrd_key
= format
!("{}/read_ios", rrd_prefix
);
741 rrd_update_derive(&rrd_key
, stat
.read_ios
as f64);
742 let rrd_key
= format
!("{}/read_bytes", rrd_prefix
);
743 rrd_update_derive(&rrd_key
, (stat
.read_sectors
*512) as f64);
745 let rrd_key
= format
!("{}/write_ios", rrd_prefix
);
746 rrd_update_derive(&rrd_key
, stat
.write_ios
as f64);
747 let rrd_key
= format
!("{}/write_bytes", rrd_prefix
);
748 rrd_update_derive(&rrd_key
, (stat
.write_sectors
*512) as f64);
750 let rrd_key
= format
!("{}/io_ticks", rrd_prefix
);
751 rrd_update_derive(&rrd_key
, (stat
.io_ticks
as f64)/1000.0);
756 eprintln
!("disk_manager mount_info() failed - {}", err
);
761 // Returns (total, used, avail)
762 fn disk_usage(path
: &std
::path
::Path
) -> Result
<(u64, u64, u64), Error
> {
764 let mut stat
: libc
::statfs64
= unsafe { std::mem::zeroed() }
;
768 let res
= path
.with_nix_path(|cstr
| unsafe { libc::statfs64(cstr.as_ptr(), &mut stat) }
)?
;
769 nix
::errno
::Errno
::result(res
)?
;
771 let bsize
= stat
.f_bsize
as u64;
773 Ok((stat
.f_blocks
*bsize
, (stat
.f_blocks
-stat
.f_bfree
)*bsize
, stat
.f_bavail
*bsize
))
776 // Returns (fs_type, device, mount_source)
777 pub fn find_mounted_device(
778 mountinfo
: &MountInfo
,
779 path
: &std
::path
::Path
,
780 ) -> Option
<(String
, Device
, Option
<OsString
>)> {
782 let mut result
= None
;
783 let mut match_len
= 0;
785 let root_path
= std
::path
::Path
::new("/");
786 for (_id
, entry
) in mountinfo
{
787 if entry
.root
== root_path
&& path
.starts_with(&entry
.mount_point
) {
788 let len
= entry
.mount_point
.as_path().as_os_str().len();
791 result
= Some((entry
.fs_type
.clone(), entry
.device
, entry
.mount_source
.clone()));