2 use std
::path
::{Path, PathBuf}
;
3 use std
::os
::unix
::io
::AsRawFd
;
5 use anyhow
::{bail, format_err, Error}
;
8 use openssl
::ssl
::{SslMethod, SslAcceptor, SslFiletype}
;
10 use proxmox
::try_block
;
11 use proxmox
::api
::RpcEnvironmentType
;
24 rotate_task_log_archive
,
26 tools
::systemd
::time
::{
33 use proxmox_backup
::api2
::types
::Authid
;
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
::{
46 PROXMOX_BACKUP_TCP_KEEPALIVE_TIME
,
50 use proxmox_backup
::api2
::pull
::do_sync_job
;
51 use proxmox_backup
::server
::do_verification_job
;
52 use proxmox_backup
::server
::do_prune_job
;
54 fn main() -> Result
<(), Error
> {
55 proxmox_backup
::tools
::setup_safe_path_env();
57 let backup_uid
= proxmox_backup
::backup
::backup_user()?
.uid
;
58 let backup_gid
= proxmox_backup
::backup
::backup_group()?
.gid
;
59 let running_uid
= nix
::unistd
::Uid
::effective();
60 let running_gid
= nix
::unistd
::Gid
::effective();
62 if running_uid
!= backup_uid
|| running_gid
!= backup_gid
{
63 bail
!("proxy not running as backup user or group (got uid {} gid {})", running_uid
, running_gid
);
66 proxmox_backup
::tools
::runtime
::main(run())
69 async
fn run() -> Result
<(), Error
> {
70 if let Err(err
) = syslog
::init(
71 syslog
::Facility
::LOG_DAEMON
,
72 log
::LevelFilter
::Info
,
73 Some("proxmox-backup-proxy")) {
74 bail
!("unable to inititialize syslog - {}", err
);
77 let _
= public_auth_key(); // load with lazy_static
78 let _
= csrf_secret(); // load with lazy_static
80 let mut config
= ApiConfig
::new(
81 buildcfg
::JS_DIR
, &proxmox_backup
::api2
::ROUTER
, RpcEnvironmentType
::PUBLIC
)?
;
83 config
.add_alias("novnc", "/usr/share/novnc-pve");
84 config
.add_alias("extjs", "/usr/share/javascript/extjs");
85 config
.add_alias("fontawesome", "/usr/share/fonts-font-awesome");
86 config
.add_alias("xtermjs", "/usr/share/pve-xtermjs");
87 config
.add_alias("locale", "/usr/share/pbs-i18n");
88 config
.add_alias("widgettoolkit", "/usr/share/javascript/proxmox-widget-toolkit");
89 config
.add_alias("css", "/usr/share/javascript/proxmox-backup/css");
90 config
.add_alias("docs", "/usr/share/doc/proxmox-backup/html");
92 let mut indexpath
= PathBuf
::from(buildcfg
::JS_DIR
);
93 indexpath
.push("index.hbs");
94 config
.register_template("index", &indexpath
)?
;
95 config
.register_template("console", "/usr/share/pve-xtermjs/index.html.hbs")?
;
97 let mut commando_sock
= server
::CommandoSocket
::new(server
::our_ctrl_sock());
99 config
.enable_file_log(buildcfg
::API_ACCESS_LOG_FN
)?
;
101 let rest_server
= RestServer
::new(config
);
103 //openssl req -x509 -newkey rsa:4096 -keyout /etc/proxmox-backup/proxy.key -out /etc/proxmox-backup/proxy.pem -nodes
104 let key_path
= configdir
!("/proxy.key");
105 let cert_path
= configdir
!("/proxy.pem");
107 let mut acceptor
= SslAcceptor
::mozilla_intermediate_v5(SslMethod
::tls()).unwrap();
108 acceptor
.set_private_key_file(key_path
, SslFiletype
::PEM
)
109 .map_err(|err
| format_err
!("unable to read proxy key {} - {}", key_path
, err
))?
;
110 acceptor
.set_certificate_chain_file(cert_path
)
111 .map_err(|err
| format_err
!("unable to read proxy cert {} - {}", cert_path
, err
))?
;
112 acceptor
.check_private_key().unwrap();
114 let acceptor
= Arc
::new(acceptor
.build());
116 let server
= daemon
::create_daemon(
117 ([0,0,0,0,0,0,0,0], 8007).into(),
119 let connections
= proxmox_backup
::tools
::async_io
::StaticIncoming
::from(listener
)
120 .map_err(Error
::from
)
121 .try_filter_map(move |(sock
, _addr
)| {
122 let acceptor
= Arc
::clone(&acceptor
);
124 sock
.set_nodelay(true).unwrap();
126 let _
= set_tcp_keepalive(sock
.as_raw_fd(), PROXMOX_BACKUP_TCP_KEEPALIVE_TIME
);
128 Ok(tokio_openssl
::accept(&acceptor
, sock
)
130 .ok() // handshake errors aren't be fatal, so return None to filter
134 let connections
= proxmox_backup
::tools
::async_io
::HyperAccept(connections
);
137 .and_then(|_
| hyper
::Server
::builder(connections
)
139 .with_graceful_shutdown(server
::shutdown_future())
140 .map_err(Error
::from
)
142 .map_err(|err
| eprintln
!("server error: {}", err
))
148 daemon
::systemd_notify(daemon
::SystemdNotify
::Ready
)?
;
150 let init_result
: Result
<(), Error
> = try_block
!({
151 server
::register_task_control_commands(&mut commando_sock
)?
;
152 commando_sock
.spawn()?
;
153 server
::server_state_init()?
;
157 if let Err(err
) = init_result
{
158 bail
!("unable to start daemon - {}", err
);
161 start_task_scheduler();
162 start_stat_generator();
165 log
::info
!("server shutting down, waiting for active workers to complete");
166 proxmox_backup
::server
::last_worker_future().await?
;
167 log
::info
!("done - exit server");
172 fn start_stat_generator() {
173 let abort_future
= server
::shutdown_future();
174 let future
= Box
::pin(run_stat_generator());
175 let task
= futures
::future
::select(future
, abort_future
);
176 tokio
::spawn(task
.map(|_
| ()));
179 fn start_task_scheduler() {
180 let abort_future
= server
::shutdown_future();
181 let future
= Box
::pin(run_task_scheduler());
182 let task
= futures
::future
::select(future
, abort_future
);
183 tokio
::spawn(task
.map(|_
| ()));
186 use std
::time
::{SystemTime, Instant, Duration, UNIX_EPOCH}
;
188 fn next_minute() -> Result
<Instant
, Error
> {
189 let now
= SystemTime
::now();
190 let epoch_now
= now
.duration_since(UNIX_EPOCH
)?
;
191 let epoch_next
= Duration
::from_secs((epoch_now
.as_secs()/60 + 1)*60);
192 Ok(Instant
::now() + epoch_next
- epoch_now
)
195 async
fn run_task_scheduler() {
197 let mut count
: usize = 0;
202 let delay_target
= match next_minute() { // try to run very minute
205 eprintln
!("task scheduler: compute next minute failed - {}", err
);
206 tokio
::time
::delay_until(tokio
::time
::Instant
::from_std(Instant
::now() + Duration
::from_secs(60))).await
;
211 if count
> 2 { // wait 1..2 minutes before starting
212 match schedule_tasks().catch_unwind().await
{
214 match panic
.downcast
::<&str>() {
216 eprintln
!("task scheduler panic: {}", msg
);
219 eprintln
!("task scheduler panic - unknown type");
224 eprintln
!("task scheduler failed - {:?}", err
);
230 tokio
::time
::delay_until(tokio
::time
::Instant
::from_std(delay_target
)).await
;
234 async
fn schedule_tasks() -> Result
<(), Error
> {
236 schedule_datastore_garbage_collection().await
;
237 schedule_datastore_prune().await
;
238 schedule_datastore_sync_jobs().await
;
239 schedule_datastore_verify_jobs().await
;
240 schedule_task_log_rotate().await
;
245 async
fn schedule_datastore_garbage_collection() {
247 use proxmox_backup
::config
::{
254 let config
= match datastore
::config() {
256 eprintln
!("unable to read datastore config - {}", err
);
259 Ok((config
, _digest
)) => config
,
262 for (store
, (_
, store_config
)) in config
.sections
{
263 let datastore
= match DataStore
::lookup_datastore(&store
) {
264 Ok(datastore
) => datastore
,
266 eprintln
!("lookup_datastore failed - {}", err
);
271 let store_config
: DataStoreConfig
= match serde_json
::from_value(store_config
) {
274 eprintln
!("datastore config from_value failed - {}", err
);
279 let event_str
= match store_config
.gc_schedule
{
280 Some(event_str
) => event_str
,
284 let event
= match parse_calendar_event(&event_str
) {
287 eprintln
!("unable to parse schedule '{}' - {}", event_str
, err
);
292 if datastore
.garbage_collection_running() { continue; }
294 let worker_type
= "garbage_collection";
296 let stat
= datastore
.last_gc_status();
297 let last
= if let Some(upid_str
) = stat
.upid
{
298 match upid_str
.parse
::<UPID
>() {
299 Ok(upid
) => upid
.starttime
,
301 eprintln
!("unable to parse upid '{}' - {}", upid_str
, err
);
306 match jobstate
::last_run_time(worker_type
, &store
) {
309 eprintln
!("could not get last run time of {} {}: {}", worker_type
, store
, err
);
315 let next
= match compute_next_event(&event
, last
, false) {
316 Ok(Some(next
)) => next
,
317 Ok(None
) => continue,
319 eprintln
!("compute_next_event for '{}' failed - {}", event_str
, err
);
324 let now
= proxmox
::tools
::time
::epoch_i64();
326 if next
> now { continue; }
328 let job
= match Job
::new(worker_type
, &store
) {
330 Err(_
) => continue, // could not get lock
333 let auth_id
= Authid
::backup_auth_id();
335 if let Err(err
) = crate::server
::do_garbage_collection_job(job
, datastore
, auth_id
, Some(event_str
)) {
336 eprintln
!("unable to start garbage collection job on datastore {} - {}", store
, err
);
341 async
fn schedule_datastore_prune() {
343 use proxmox_backup
::{
353 let config
= match datastore
::config() {
355 eprintln
!("unable to read datastore config - {}", err
);
358 Ok((config
, _digest
)) => config
,
361 for (store
, (_
, store_config
)) in config
.sections
{
363 let store_config
: DataStoreConfig
= match serde_json
::from_value(store_config
) {
366 eprintln
!("datastore '{}' config from_value failed - {}", store
, err
);
371 let event_str
= match store_config
.prune_schedule
{
372 Some(event_str
) => event_str
,
376 let prune_options
= PruneOptions
{
377 keep_last
: store_config
.keep_last
,
378 keep_hourly
: store_config
.keep_hourly
,
379 keep_daily
: store_config
.keep_daily
,
380 keep_weekly
: store_config
.keep_weekly
,
381 keep_monthly
: store_config
.keep_monthly
,
382 keep_yearly
: store_config
.keep_yearly
,
385 if !prune_options
.keeps_something() { // no prune settings - keep all
389 let worker_type
= "prune";
390 if check_schedule(worker_type
, &event_str
, &store
) {
391 let job
= match Job
::new(worker_type
, &store
) {
393 Err(_
) => continue, // could not get lock
396 let auth_id
= Authid
::backup_auth_id().clone();
397 if let Err(err
) = do_prune_job(job
, prune_options
, store
.clone(), &auth_id
, Some(event_str
)) {
398 eprintln
!("unable to start datastore prune job {} - {}", &store
, err
);
404 async
fn schedule_datastore_sync_jobs() {
406 use proxmox_backup
::config
::sync
::{
411 let config
= match sync
::config() {
413 eprintln
!("unable to read sync job config - {}", err
);
416 Ok((config
, _digest
)) => config
,
419 for (job_id
, (_
, job_config
)) in config
.sections
{
420 let job_config
: SyncJobConfig
= match serde_json
::from_value(job_config
) {
423 eprintln
!("sync job config from_value failed - {}", err
);
428 let event_str
= match job_config
.schedule
{
429 Some(ref event_str
) => event_str
.clone(),
433 let worker_type
= "syncjob";
434 if check_schedule(worker_type
, &event_str
, &job_id
) {
435 let job
= match Job
::new(worker_type
, &job_id
) {
437 Err(_
) => continue, // could not get lock
440 let auth_id
= Authid
::backup_auth_id().clone();
441 if let Err(err
) = do_sync_job(job
, job_config
, &auth_id
, Some(event_str
)) {
442 eprintln
!("unable to start datastore sync job {} - {}", &job_id
, err
);
448 async
fn schedule_datastore_verify_jobs() {
450 use proxmox_backup
::config
::verify
::{
452 VerificationJobConfig
,
455 let config
= match verify
::config() {
457 eprintln
!("unable to read verification job config - {}", err
);
460 Ok((config
, _digest
)) => config
,
462 for (job_id
, (_
, job_config
)) in config
.sections
{
463 let job_config
: VerificationJobConfig
= match serde_json
::from_value(job_config
) {
466 eprintln
!("verification job config from_value failed - {}", err
);
470 let event_str
= match job_config
.schedule
{
471 Some(ref event_str
) => event_str
.clone(),
475 let worker_type
= "verificationjob";
476 let auth_id
= Authid
::backup_auth_id().clone();
477 if check_schedule(worker_type
, &event_str
, &job_id
) {
478 let job
= match Job
::new(&worker_type
, &job_id
) {
480 Err(_
) => continue, // could not get lock
482 if let Err(err
) = do_verification_job(job
, job_config
, &auth_id
, Some(event_str
)) {
483 eprintln
!("unable to start datastore verification job {} - {}", &job_id
, err
);
489 async
fn schedule_task_log_rotate() {
491 let worker_type
= "logrotate";
492 let job_id
= "task_archive";
494 // schedule daily at 00:00 like normal logrotate
495 let schedule
= "00:00";
497 if !check_schedule(worker_type
, schedule
, job_id
) {
498 // if we never ran the rotation, schedule instantly
499 match jobstate
::JobState
::load(worker_type
, job_id
) {
500 Ok(state
) => match state
{
501 jobstate
::JobState
::Created { .. }
=> {}
,
508 let mut job
= match Job
::new(worker_type
, job_id
) {
510 Err(_
) => return, // could not get lock
513 if let Err(err
) = WorkerTask
::new_thread(
515 Some(job_id
.to_string()),
516 Authid
::backup_auth_id().clone(),
519 job
.start(&worker
.upid().to_string())?
;
520 worker
.log(format
!("starting task log rotation"));
522 let result
= try_block
!({
523 // rotate task log archive
524 let max_size
= 500000; // a normal entry has about 100b, so ~ 5000 entries/file
525 let max_files
= 20; // times twenty files gives at least 100000 task entries
526 let has_rotated
= rotate_task_log_archive(max_size
, true, Some(max_files
))?
;
528 worker
.log(format
!("task log archive was rotated"));
530 worker
.log(format
!("task log archive was not rotated"));
536 let status
= worker
.create_state(&result
);
538 if let Err(err
) = job
.finish(status
) {
539 eprintln
!("could not finish job state for {}: {}", worker_type
, err
);
545 eprintln
!("unable to start task log rotation: {}", err
);
550 async
fn run_stat_generator() {
555 let save
= if count
>= 6 { count = 0; true }
else { false }
;
557 let delay_target
= Instant
::now() + Duration
::from_secs(10);
559 generate_host_stats(save
).await
;
561 tokio
::time
::delay_until(tokio
::time
::Instant
::from_std(delay_target
)).await
;
567 fn rrd_update_gauge(name
: &str, value
: f64, save
: bool
) {
568 use proxmox_backup
::rrd
;
569 if let Err(err
) = rrd
::update_value(name
, value
, rrd
::DST
::Gauge
, save
) {
570 eprintln
!("rrd::update_value '{}' failed - {}", name
, err
);
574 fn rrd_update_derive(name
: &str, value
: f64, save
: bool
) {
575 use proxmox_backup
::rrd
;
576 if let Err(err
) = rrd
::update_value(name
, value
, rrd
::DST
::Derive
, save
) {
577 eprintln
!("rrd::update_value '{}' failed - {}", name
, err
);
581 async
fn generate_host_stats(save
: bool
) {
582 use proxmox
::sys
::linux
::procfs
::{
583 read_meminfo
, read_proc_stat
, read_proc_net_dev
, read_loadavg
};
584 use proxmox_backup
::config
::datastore
;
587 proxmox_backup
::tools
::runtime
::block_in_place(move || {
589 match read_proc_stat() {
591 rrd_update_gauge("host/cpu", stat
.cpu
, save
);
592 rrd_update_gauge("host/iowait", stat
.iowait_percent
, save
);
595 eprintln
!("read_proc_stat failed - {}", err
);
599 match read_meminfo() {
601 rrd_update_gauge("host/memtotal", meminfo
.memtotal
as f64, save
);
602 rrd_update_gauge("host/memused", meminfo
.memused
as f64, save
);
603 rrd_update_gauge("host/swaptotal", meminfo
.swaptotal
as f64, save
);
604 rrd_update_gauge("host/swapused", meminfo
.swapused
as f64, save
);
607 eprintln
!("read_meminfo failed - {}", err
);
611 match read_proc_net_dev() {
613 use proxmox_backup
::config
::network
::is_physical_nic
;
617 if !is_physical_nic(&item
.device
) { continue; }
618 netin
+= item
.receive
;
621 rrd_update_derive("host/netin", netin
as f64, save
);
622 rrd_update_derive("host/netout", netout
as f64, save
);
625 eprintln
!("read_prox_net_dev failed - {}", err
);
629 match read_loadavg() {
631 rrd_update_gauge("host/loadavg", loadavg
.0 as f64, save
);
634 eprintln
!("read_loadavg failed - {}", err
);
638 let disk_manager
= DiskManage
::new();
640 gather_disk_stats(disk_manager
.clone(), Path
::new("/"), "host", save
);
642 match datastore
::config() {
644 let datastore_list
: Vec
<datastore
::DataStoreConfig
> =
645 config
.convert_to_typed_array("datastore").unwrap_or(Vec
::new());
647 for config
in datastore_list
{
649 let rrd_prefix
= format
!("datastore/{}", config
.name
);
650 let path
= std
::path
::Path
::new(&config
.path
);
651 gather_disk_stats(disk_manager
.clone(), path
, &rrd_prefix
, save
);
655 eprintln
!("read datastore config failed - {}", err
);
662 fn check_schedule(worker_type
: &str, event_str
: &str, id
: &str) -> bool
{
663 let event
= match parse_calendar_event(event_str
) {
666 eprintln
!("unable to parse schedule '{}' - {}", event_str
, err
);
671 let last
= match jobstate
::last_run_time(worker_type
, &id
) {
674 eprintln
!("could not get last run time of {} {}: {}", worker_type
, id
, err
);
679 let next
= match compute_next_event(&event
, last
, false) {
680 Ok(Some(next
)) => next
,
681 Ok(None
) => return false,
683 eprintln
!("compute_next_event for '{}' failed - {}", event_str
, err
);
688 let now
= proxmox
::tools
::time
::epoch_i64();
692 fn gather_disk_stats(disk_manager
: Arc
<DiskManage
>, path
: &Path
, rrd_prefix
: &str, save
: bool
) {
694 match proxmox_backup
::tools
::disks
::disk_usage(path
) {
696 let rrd_key
= format
!("{}/total", rrd_prefix
);
697 rrd_update_gauge(&rrd_key
, status
.total
as f64, save
);
698 let rrd_key
= format
!("{}/used", rrd_prefix
);
699 rrd_update_gauge(&rrd_key
, status
.used
as f64, save
);
702 eprintln
!("read disk_usage on {:?} failed - {}", path
, err
);
706 match disk_manager
.find_mounted_device(path
) {
708 Ok(Some((fs_type
, device
, source
))) => {
709 let mut device_stat
= None
;
710 match fs_type
.as_str() {
712 if let Some(pool
) = source
{
713 match zfs_pool_stats(&pool
) {
714 Ok(stat
) => device_stat
= stat
,
715 Err(err
) => eprintln
!("zfs_pool_stats({:?}) failed - {}", pool
, err
),
720 if let Ok(disk
) = disk_manager
.clone().disk_by_dev_num(device
.into_dev_t()) {
721 match disk
.read_stat() {
722 Ok(stat
) => device_stat
= stat
,
723 Err(err
) => eprintln
!("disk.read_stat {:?} failed - {}", path
, err
),
728 if let Some(stat
) = device_stat
{
729 let rrd_key
= format
!("{}/read_ios", rrd_prefix
);
730 rrd_update_derive(&rrd_key
, stat
.read_ios
as f64, save
);
731 let rrd_key
= format
!("{}/read_bytes", rrd_prefix
);
732 rrd_update_derive(&rrd_key
, (stat
.read_sectors
*512) as f64, save
);
734 let rrd_key
= format
!("{}/write_ios", rrd_prefix
);
735 rrd_update_derive(&rrd_key
, stat
.write_ios
as f64, save
);
736 let rrd_key
= format
!("{}/write_bytes", rrd_prefix
);
737 rrd_update_derive(&rrd_key
, (stat
.write_sectors
*512) as f64, save
);
739 let rrd_key
= format
!("{}/io_ticks", rrd_prefix
);
740 rrd_update_derive(&rrd_key
, (stat
.io_ticks
as f64)/1000.0, save
);
744 eprintln
!("find_mounted_device failed - {}", err
);