1 use std
::sync
::{Mutex, Arc}
;
2 use std
::path
::{Path, PathBuf}
;
3 use std
::os
::unix
::io
::AsRawFd
;
4 use std
::future
::Future
;
7 use anyhow
::{bail, format_err, Error}
;
9 use http
::request
::Parts
;
11 use hyper
::{Body, StatusCode}
;
13 use url
::form_urlencoded
;
15 use openssl
::ssl
::{SslMethod, SslAcceptor, SslFiletype}
;
16 use tokio_stream
::wrappers
::ReceiverStream
;
17 use serde_json
::{json, Value}
;
18 use http
::{Method, HeaderMap}
;
20 use proxmox
::sys
::linux
::socket
::set_tcp_keepalive
;
21 use proxmox
::tools
::fs
::CreateOptions
;
22 use proxmox_lang
::try_block
;
23 use proxmox_router
::{RpcEnvironment, RpcEnvironmentType, UserInformation}
;
25 use pbs_tools
::{task_log, task_warn}
;
26 use pbs_datastore
::DataStore
;
29 use proxmox_rest_server
::{
30 rotate_task_log_archive
, extract_cookie
, AuthError
, ApiConfig
, RestServer
, RestEnvironment
,
31 ServerAdapter
, WorkerTask
, cleanup_old_tasks
,
45 use pbs_buildcfg
::configdir
;
46 use proxmox_systemd
::time
::{compute_next_event, parse_calendar_event}
;
47 use pbs_tools
::logrotate
::LogRotate
;
50 Authid
, TapeBackupJobConfig
, VerificationJobConfig
, SyncJobConfig
, DataStoreConfig
,
54 use proxmox_rest_server
::daemon
;
56 use proxmox_backup
::server
;
57 use proxmox_backup
::auth_helpers
::*;
58 use proxmox_backup
::tools
::{
59 PROXMOX_BACKUP_TCP_KEEPALIVE_TIME
,
63 get_pool_from_dataset
,
68 use proxmox_backup
::api2
::pull
::do_sync_job
;
69 use proxmox_backup
::api2
::tape
::backup
::do_tape_backup_job
;
70 use proxmox_backup
::server
::do_verification_job
;
71 use proxmox_backup
::server
::do_prune_job
;
73 fn main() -> Result
<(), Error
> {
74 proxmox_backup
::tools
::setup_safe_path_env();
76 let backup_uid
= pbs_config
::backup_user()?
.uid
;
77 let backup_gid
= pbs_config
::backup_group()?
.gid
;
78 let running_uid
= nix
::unistd
::Uid
::effective();
79 let running_gid
= nix
::unistd
::Gid
::effective();
81 if running_uid
!= backup_uid
|| running_gid
!= backup_gid
{
82 bail
!("proxy not running as backup user or group (got uid {} gid {})", running_uid
, running_gid
);
85 pbs_runtime
::main(run())
89 struct ProxmoxBackupProxyAdapter
;
91 impl ServerAdapter
for ProxmoxBackupProxyAdapter
{
97 ) -> Pin
<Box
<dyn Future
<Output
= Response
<Body
>> + Send
>> {
98 Box
::pin(get_index_future(env
, parts
))
103 headers
: &'a HeaderMap
,
105 ) -> Pin
<Box
<dyn Future
<Output
= Result
<(String
, Box
<dyn UserInformation
+ Sync
+ Send
>), AuthError
>> + Send
+ 'a
>> {
106 Box
::pin(async
move {
107 check_pbs_auth(headers
, method
).await
112 fn extract_lang_header(headers
: &http
::HeaderMap
) -> Option
<String
> {
113 if let Some(Ok(cookie
)) = headers
.get("COOKIE").map(|v
| v
.to_str()) {
114 return extract_cookie(cookie
, "PBSLangCookie");
119 async
fn get_index_future(
120 env
: RestEnvironment
,
122 ) -> Response
<Body
> {
124 let auth_id
= env
.get_auth_id();
125 let api
= env
.api_config();
126 let language
= extract_lang_header(&parts
.headers
);
128 // fixme: make all IO async
130 let (userid
, csrf_token
) = match auth_id
{
132 let auth_id
= auth_id
.parse
::<Authid
>();
134 Ok(auth_id
) if !auth_id
.is_token() => {
135 let userid
= auth_id
.user().clone();
136 let new_csrf_token
= assemble_csrf_prevention_token(csrf_secret(), &userid
);
137 (Some(userid
), Some(new_csrf_token
))
142 None
=> (None
, None
),
145 let nodename
= proxmox
::tools
::nodename();
146 let user
= userid
.as_ref().map(|u
| u
.as_str()).unwrap_or("");
148 let csrf_token
= csrf_token
.unwrap_or_else(|| String
::from(""));
150 let mut debug
= false;
151 let mut template_file
= "index";
153 if let Some(query_str
) = parts
.uri
.query() {
154 for (k
, v
) in form_urlencoded
::parse(query_str
.as_bytes()).into_owned() {
155 if k
== "debug" && v
!= "0" && v
!= "false" {
157 } else if k
== "console" {
158 template_file
= "console";
163 let mut lang
= String
::from("");
164 if let Some(language
) = language
{
165 if Path
::new(&format
!("/usr/share/pbs-i18n/pbs-lang-{}.js", language
)).exists() {
171 "NodeName": nodename
,
173 "CSRFPreventionToken": csrf_token
,
178 let (ct
, index
) = match api
.render_template(template_file
, &data
) {
179 Ok(index
) => ("text/html", index
),
180 Err(err
) => ("text/plain", format
!("Error rendering template: {}", err
)),
183 let mut resp
= Response
::builder()
184 .status(StatusCode
::OK
)
185 .header(header
::CONTENT_TYPE
, ct
)
189 if let Some(userid
) = userid
{
190 resp
.extensions_mut().insert(Authid
::from((userid
, None
)));
196 async
fn run() -> Result
<(), Error
> {
197 if let Err(err
) = syslog
::init(
198 syslog
::Facility
::LOG_DAEMON
,
199 log
::LevelFilter
::Info
,
200 Some("proxmox-backup-proxy")) {
201 bail
!("unable to inititialize syslog - {}", err
);
204 // Note: To debug early connection error use
205 // PROXMOX_DEBUG=1 ./target/release/proxmox-backup-proxy
206 let debug
= std
::env
::var("PROXMOX_DEBUG").is_ok();
208 let _
= public_auth_key(); // load with lazy_static
209 let _
= csrf_secret(); // load with lazy_static
211 let mut config
= ApiConfig
::new(
212 pbs_buildcfg
::JS_DIR
,
213 &proxmox_backup
::api2
::ROUTER
,
214 RpcEnvironmentType
::PUBLIC
,
215 ProxmoxBackupProxyAdapter
,
218 config
.add_alias("novnc", "/usr/share/novnc-pve");
219 config
.add_alias("extjs", "/usr/share/javascript/extjs");
220 config
.add_alias("qrcodejs", "/usr/share/javascript/qrcodejs");
221 config
.add_alias("fontawesome", "/usr/share/fonts-font-awesome");
222 config
.add_alias("xtermjs", "/usr/share/pve-xtermjs");
223 config
.add_alias("locale", "/usr/share/pbs-i18n");
224 config
.add_alias("widgettoolkit", "/usr/share/javascript/proxmox-widget-toolkit");
225 config
.add_alias("docs", "/usr/share/doc/proxmox-backup/html");
227 let mut indexpath
= PathBuf
::from(pbs_buildcfg
::JS_DIR
);
228 indexpath
.push("index.hbs");
229 config
.register_template("index", &indexpath
)?
;
230 config
.register_template("console", "/usr/share/pve-xtermjs/index.html.hbs")?
;
232 let backup_user
= pbs_config
::backup_user()?
;
233 let mut commando_sock
= proxmox_rest_server
::CommandSocket
::new(proxmox_rest_server
::our_ctrl_sock(), backup_user
.gid
);
235 let dir_opts
= CreateOptions
::new().owner(backup_user
.uid
).group(backup_user
.gid
);
236 let file_opts
= CreateOptions
::new().owner(backup_user
.uid
).group(backup_user
.gid
);
238 config
.enable_access_log(
239 pbs_buildcfg
::API_ACCESS_LOG_FN
,
240 Some(dir_opts
.clone()),
241 Some(file_opts
.clone()),
245 config
.enable_auth_log(
246 pbs_buildcfg
::API_AUTH_LOG_FN
,
247 Some(dir_opts
.clone()),
248 Some(file_opts
.clone()),
252 let rest_server
= RestServer
::new(config
);
253 proxmox_rest_server
::init_worker_tasks(pbs_buildcfg
::PROXMOX_BACKUP_LOG_DIR_M
!().into(), file_opts
.clone())?
;
255 //openssl req -x509 -newkey rsa:4096 -keyout /etc/proxmox-backup/proxy.key -out /etc/proxmox-backup/proxy.pem -nodes
257 // we build the initial acceptor here as we cannot start if this fails
258 let acceptor
= make_tls_acceptor()?
;
259 let acceptor
= Arc
::new(Mutex
::new(acceptor
));
261 // to renew the acceptor we just add a command-socket handler
262 commando_sock
.register_command(
263 "reload-certificate".to_string(),
265 let acceptor
= Arc
::clone(&acceptor
);
266 move |_value
| -> Result
<_
, Error
> {
267 log
::info
!("reloading certificate");
268 match make_tls_acceptor() {
269 Err(err
) => log
::error
!("error reloading certificate: {}", err
),
270 Ok(new_acceptor
) => {
271 let mut guard
= acceptor
.lock().unwrap();
272 *guard
= new_acceptor
;
280 // to remove references for not configured datastores
281 commando_sock
.register_command(
282 "datastore-removed".to_string(),
284 if let Err(err
) = DataStore
::remove_unused_datastores() {
285 log
::error
!("could not refresh datastores: {}", err
);
291 let server
= daemon
::create_daemon(
292 ([0,0,0,0,0,0,0,0], 8007).into(),
295 let connections
= accept_connections(listener
, acceptor
, debug
);
296 let connections
= hyper
::server
::accept
::from_stream(ReceiverStream
::new(connections
));
299 daemon
::systemd_notify(daemon
::SystemdNotify
::Ready
)?
;
301 hyper
::Server
::builder(connections
)
303 .with_graceful_shutdown(proxmox_rest_server
::shutdown_future())
304 .map_err(Error
::from
)
310 proxmox_rest_server
::write_pid(pbs_buildcfg
::PROXMOX_BACKUP_PROXY_PID_FN
)?
;
312 let init_result
: Result
<(), Error
> = try_block
!({
313 proxmox_rest_server
::register_task_control_commands(&mut commando_sock
)?
;
314 commando_sock
.spawn()?
;
315 proxmox_rest_server
::catch_shutdown_signal()?
;
316 proxmox_rest_server
::catch_reload_signal()?
;
320 if let Err(err
) = init_result
{
321 bail
!("unable to start daemon - {}", err
);
324 start_task_scheduler();
325 start_stat_generator();
328 log
::info
!("server shutting down, waiting for active workers to complete");
329 proxmox_rest_server
::last_worker_future().await?
;
330 log
::info
!("done - exit server");
335 fn make_tls_acceptor() -> Result
<SslAcceptor
, Error
> {
336 let key_path
= configdir
!("/proxy.key");
337 let cert_path
= configdir
!("/proxy.pem");
339 let mut acceptor
= SslAcceptor
::mozilla_intermediate_v5(SslMethod
::tls()).unwrap();
340 acceptor
.set_private_key_file(key_path
, SslFiletype
::PEM
)
341 .map_err(|err
| format_err
!("unable to read proxy key {} - {}", key_path
, err
))?
;
342 acceptor
.set_certificate_chain_file(cert_path
)
343 .map_err(|err
| format_err
!("unable to read proxy cert {} - {}", cert_path
, err
))?
;
344 acceptor
.check_private_key().unwrap();
349 type ClientStreamResult
=
350 Result
<std
::pin
::Pin
<Box
<tokio_openssl
::SslStream
<tokio
::net
::TcpStream
>>>, Error
>;
351 const MAX_PENDING_ACCEPTS
: usize = 1024;
353 fn accept_connections(
354 listener
: tokio
::net
::TcpListener
,
355 acceptor
: Arc
<Mutex
<openssl
::ssl
::SslAcceptor
>>,
357 ) -> tokio
::sync
::mpsc
::Receiver
<ClientStreamResult
> {
359 let (sender
, receiver
) = tokio
::sync
::mpsc
::channel(MAX_PENDING_ACCEPTS
);
361 tokio
::spawn(accept_connection(listener
, acceptor
, debug
, sender
));
366 async
fn accept_connection(
367 listener
: tokio
::net
::TcpListener
,
368 acceptor
: Arc
<Mutex
<openssl
::ssl
::SslAcceptor
>>,
370 sender
: tokio
::sync
::mpsc
::Sender
<ClientStreamResult
>,
372 let accept_counter
= Arc
::new(());
375 let (sock
, _addr
) = match listener
.accept().await
{
378 eprintln
!("error accepting tcp connection: {}", err
);
383 sock
.set_nodelay(true).unwrap();
384 let _
= set_tcp_keepalive(sock
.as_raw_fd(), PROXMOX_BACKUP_TCP_KEEPALIVE_TIME
);
386 let ssl
= { // limit acceptor_guard scope
387 // Acceptor can be reloaded using the command socket "reload-certificate" command
388 let acceptor_guard
= acceptor
.lock().unwrap();
390 match openssl
::ssl
::Ssl
::new(acceptor_guard
.context()) {
393 eprintln
!("failed to create Ssl object from Acceptor context - {}", err
);
399 let stream
= match tokio_openssl
::SslStream
::new(ssl
, sock
) {
400 Ok(stream
) => stream
,
402 eprintln
!("failed to create SslStream using ssl and connection socket - {}", err
);
407 let mut stream
= Box
::pin(stream
);
408 let sender
= sender
.clone();
410 if Arc
::strong_count(&accept_counter
) > MAX_PENDING_ACCEPTS
{
411 eprintln
!("connection rejected - to many open connections");
415 let accept_counter
= Arc
::clone(&accept_counter
);
416 tokio
::spawn(async
move {
417 let accept_future
= tokio
::time
::timeout(
418 Duration
::new(10, 0), stream
.as_mut().accept());
420 let result
= accept_future
.await
;
424 if sender
.send(Ok(stream
)).await
.is_err() && debug
{
425 eprintln
!("detect closed connection channel");
430 eprintln
!("https handshake failed - {}", err
);
435 eprintln
!("https handshake timeout");
440 drop(accept_counter
); // decrease reference count
445 fn start_stat_generator() {
446 let abort_future
= proxmox_rest_server
::shutdown_future();
447 let future
= Box
::pin(run_stat_generator());
448 let task
= futures
::future
::select(future
, abort_future
);
449 tokio
::spawn(task
.map(|_
| ()));
452 fn start_task_scheduler() {
453 let abort_future
= proxmox_rest_server
::shutdown_future();
454 let future
= Box
::pin(run_task_scheduler());
455 let task
= futures
::future
::select(future
, abort_future
);
456 tokio
::spawn(task
.map(|_
| ()));
459 use std
::time
::{SystemTime, Instant, Duration, UNIX_EPOCH}
;
461 fn next_minute() -> Result
<Instant
, Error
> {
462 let now
= SystemTime
::now();
463 let epoch_now
= now
.duration_since(UNIX_EPOCH
)?
;
464 let epoch_next
= Duration
::from_secs((epoch_now
.as_secs()/60 + 1)*60);
465 Ok(Instant
::now() + epoch_next
- epoch_now
)
468 async
fn run_task_scheduler() {
470 let mut count
: usize = 0;
475 let delay_target
= match next_minute() { // try to run very minute
478 eprintln
!("task scheduler: compute next minute failed - {}", err
);
479 tokio
::time
::sleep_until(tokio
::time
::Instant
::from_std(Instant
::now() + Duration
::from_secs(60))).await
;
484 if count
> 2 { // wait 1..2 minutes before starting
485 match schedule_tasks().catch_unwind().await
{
487 match panic
.downcast
::<&str>() {
489 eprintln
!("task scheduler panic: {}", msg
);
492 eprintln
!("task scheduler panic - unknown type");
497 eprintln
!("task scheduler failed - {:?}", err
);
503 tokio
::time
::sleep_until(tokio
::time
::Instant
::from_std(delay_target
)).await
;
507 async
fn schedule_tasks() -> Result
<(), Error
> {
509 schedule_datastore_garbage_collection().await
;
510 schedule_datastore_prune().await
;
511 schedule_datastore_sync_jobs().await
;
512 schedule_datastore_verify_jobs().await
;
513 schedule_tape_backup_jobs().await
;
514 schedule_task_log_rotate().await
;
519 async
fn schedule_datastore_garbage_collection() {
521 let config
= match pbs_config
::datastore
::config() {
523 eprintln
!("unable to read datastore config - {}", err
);
526 Ok((config
, _digest
)) => config
,
529 for (store
, (_
, store_config
)) in config
.sections
{
530 let datastore
= match DataStore
::lookup_datastore(&store
) {
531 Ok(datastore
) => datastore
,
533 eprintln
!("lookup_datastore failed - {}", err
);
538 let store_config
: DataStoreConfig
= match serde_json
::from_value(store_config
) {
541 eprintln
!("datastore config from_value failed - {}", err
);
546 let event_str
= match store_config
.gc_schedule
{
547 Some(event_str
) => event_str
,
551 let event
= match parse_calendar_event(&event_str
) {
554 eprintln
!("unable to parse schedule '{}' - {}", event_str
, err
);
559 if datastore
.garbage_collection_running() { continue; }
561 let worker_type
= "garbage_collection";
563 let last
= match jobstate
::last_run_time(worker_type
, &store
) {
566 eprintln
!("could not get last run time of {} {}: {}", worker_type
, store
, err
);
571 let next
= match compute_next_event(&event
, last
, false) {
572 Ok(Some(next
)) => next
,
573 Ok(None
) => continue,
575 eprintln
!("compute_next_event for '{}' failed - {}", event_str
, err
);
580 let now
= proxmox_time
::epoch_i64();
582 if next
> now { continue; }
584 let job
= match Job
::new(worker_type
, &store
) {
586 Err(_
) => continue, // could not get lock
589 let auth_id
= Authid
::root_auth_id();
591 if let Err(err
) = crate::server
::do_garbage_collection_job(job
, datastore
, auth_id
, Some(event_str
), false) {
592 eprintln
!("unable to start garbage collection job on datastore {} - {}", store
, err
);
597 async
fn schedule_datastore_prune() {
599 let config
= match pbs_config
::datastore
::config() {
601 eprintln
!("unable to read datastore config - {}", err
);
604 Ok((config
, _digest
)) => config
,
607 for (store
, (_
, store_config
)) in config
.sections
{
609 let store_config
: DataStoreConfig
= match serde_json
::from_value(store_config
) {
612 eprintln
!("datastore '{}' config from_value failed - {}", store
, err
);
617 let event_str
= match store_config
.prune_schedule
{
618 Some(event_str
) => event_str
,
622 let prune_options
= PruneOptions
{
623 keep_last
: store_config
.keep_last
,
624 keep_hourly
: store_config
.keep_hourly
,
625 keep_daily
: store_config
.keep_daily
,
626 keep_weekly
: store_config
.keep_weekly
,
627 keep_monthly
: store_config
.keep_monthly
,
628 keep_yearly
: store_config
.keep_yearly
,
631 if !pbs_datastore
::prune
::keeps_something(&prune_options
) { // no prune settings - keep all
635 let worker_type
= "prune";
636 if check_schedule(worker_type
, &event_str
, &store
) {
637 let job
= match Job
::new(worker_type
, &store
) {
639 Err(_
) => continue, // could not get lock
642 let auth_id
= Authid
::root_auth_id().clone();
643 if let Err(err
) = do_prune_job(job
, prune_options
, store
.clone(), &auth_id
, Some(event_str
)) {
644 eprintln
!("unable to start datastore prune job {} - {}", &store
, err
);
650 async
fn schedule_datastore_sync_jobs() {
653 let config
= match pbs_config
::sync
::config() {
655 eprintln
!("unable to read sync job config - {}", err
);
658 Ok((config
, _digest
)) => config
,
661 for (job_id
, (_
, job_config
)) in config
.sections
{
662 let job_config
: SyncJobConfig
= match serde_json
::from_value(job_config
) {
665 eprintln
!("sync job config from_value failed - {}", err
);
670 let event_str
= match job_config
.schedule
{
671 Some(ref event_str
) => event_str
.clone(),
675 let worker_type
= "syncjob";
676 if check_schedule(worker_type
, &event_str
, &job_id
) {
677 let job
= match Job
::new(worker_type
, &job_id
) {
679 Err(_
) => continue, // could not get lock
682 let auth_id
= Authid
::root_auth_id().clone();
683 if let Err(err
) = do_sync_job(job
, job_config
, &auth_id
, Some(event_str
), false) {
684 eprintln
!("unable to start datastore sync job {} - {}", &job_id
, err
);
690 async
fn schedule_datastore_verify_jobs() {
692 let config
= match pbs_config
::verify
::config() {
694 eprintln
!("unable to read verification job config - {}", err
);
697 Ok((config
, _digest
)) => config
,
699 for (job_id
, (_
, job_config
)) in config
.sections
{
700 let job_config
: VerificationJobConfig
= match serde_json
::from_value(job_config
) {
703 eprintln
!("verification job config from_value failed - {}", err
);
707 let event_str
= match job_config
.schedule
{
708 Some(ref event_str
) => event_str
.clone(),
712 let worker_type
= "verificationjob";
713 let auth_id
= Authid
::root_auth_id().clone();
714 if check_schedule(worker_type
, &event_str
, &job_id
) {
715 let job
= match Job
::new(&worker_type
, &job_id
) {
717 Err(_
) => continue, // could not get lock
719 if let Err(err
) = do_verification_job(job
, job_config
, &auth_id
, Some(event_str
), false) {
720 eprintln
!("unable to start datastore verification job {} - {}", &job_id
, err
);
726 async
fn schedule_tape_backup_jobs() {
728 let config
= match pbs_config
::tape_job
::config() {
730 eprintln
!("unable to read tape job config - {}", err
);
733 Ok((config
, _digest
)) => config
,
735 for (job_id
, (_
, job_config
)) in config
.sections
{
736 let job_config
: TapeBackupJobConfig
= match serde_json
::from_value(job_config
) {
739 eprintln
!("tape backup job config from_value failed - {}", err
);
743 let event_str
= match job_config
.schedule
{
744 Some(ref event_str
) => event_str
.clone(),
748 let worker_type
= "tape-backup-job";
749 let auth_id
= Authid
::root_auth_id().clone();
750 if check_schedule(worker_type
, &event_str
, &job_id
) {
751 let job
= match Job
::new(&worker_type
, &job_id
) {
753 Err(_
) => continue, // could not get lock
755 if let Err(err
) = do_tape_backup_job(job
, job_config
.setup
, &auth_id
, Some(event_str
), false) {
756 eprintln
!("unable to start tape backup job {} - {}", &job_id
, err
);
763 async
fn schedule_task_log_rotate() {
765 let worker_type
= "logrotate";
766 let job_id
= "access-log_and_task-archive";
768 // schedule daily at 00:00 like normal logrotate
769 let schedule
= "00:00";
771 if !check_schedule(worker_type
, schedule
, job_id
) {
772 // if we never ran the rotation, schedule instantly
773 match jobstate
::JobState
::load(worker_type
, job_id
) {
774 Ok(state
) => match state
{
775 jobstate
::JobState
::Created { .. }
=> {}
,
782 let mut job
= match Job
::new(worker_type
, job_id
) {
784 Err(_
) => return, // could not get lock
787 if let Err(err
) = WorkerTask
::new_thread(
790 Authid
::root_auth_id().to_string(),
793 job
.start(&worker
.upid().to_string())?
;
794 task_log
!(worker
, "starting task log rotation");
796 let result
= try_block
!({
797 let max_size
= 512 * 1024 - 1; // an entry has ~ 100b, so > 5000 entries/file
798 let max_files
= 20; // times twenty files gives > 100000 task entries
799 let has_rotated
= rotate_task_log_archive(max_size
, true, Some(max_files
))?
;
801 task_log
!(worker
, "task log archive was rotated");
803 task_log
!(worker
, "task log archive was not rotated");
806 let max_size
= 32 * 1024 * 1024 - 1;
808 let mut logrotate
= LogRotate
::new(pbs_buildcfg
::API_ACCESS_LOG_FN
, true)
809 .ok_or_else(|| format_err
!("could not get API access log file names"))?
;
811 if logrotate
.rotate(max_size
, None
, Some(max_files
))?
{
812 println
!("rotated access log, telling daemons to re-open log file");
813 pbs_runtime
::block_on(command_reopen_access_logfiles())?
;
814 task_log
!(worker
, "API access log was rotated");
816 task_log
!(worker
, "API access log was not rotated");
819 let mut logrotate
= LogRotate
::new(pbs_buildcfg
::API_AUTH_LOG_FN
, true)
820 .ok_or_else(|| format_err
!("could not get API auth log file names"))?
;
822 if logrotate
.rotate(max_size
, None
, Some(max_files
))?
{
823 println
!("rotated auth log, telling daemons to re-open log file");
824 pbs_runtime
::block_on(command_reopen_auth_logfiles())?
;
825 task_log
!(worker
, "API authentication log was rotated");
827 task_log
!(worker
, "API authentication log was not rotated");
831 task_log
!(worker
, "cleaning up old task logs");
832 if let Err(err
) = cleanup_old_tasks(true) {
833 task_warn
!(worker
, "could not completely cleanup old tasks: {}", err
);
840 let status
= worker
.create_state(&result
);
842 if let Err(err
) = job
.finish(status
) {
843 eprintln
!("could not finish job state for {}: {}", worker_type
, err
);
849 eprintln
!("unable to start task log rotation: {}", err
);
854 async
fn command_reopen_access_logfiles() -> Result
<(), Error
> {
855 // only care about the most recent daemon instance for each, proxy & api, as other older ones
856 // should not respond to new requests anyway, but only finish their current one and then exit.
857 let sock
= proxmox_rest_server
::our_ctrl_sock();
858 let f1
= proxmox_rest_server
::send_raw_command(sock
, "{\"command\":\"api-access-log-reopen\"}\n");
860 let pid
= proxmox_rest_server
::read_pid(pbs_buildcfg
::PROXMOX_BACKUP_API_PID_FN
)?
;
861 let sock
= proxmox_rest_server
::ctrl_sock_from_pid(pid
);
862 let f2
= proxmox_rest_server
::send_raw_command(sock
, "{\"command\":\"api-access-log-reopen\"}\n");
864 match futures
::join
!(f1
, f2
) {
865 (Err(e1
), Err(e2
)) => Err(format_err
!("reopen commands failed, proxy: {}; api: {}", e1
, e2
)),
866 (Err(e1
), Ok(_
)) => Err(format_err
!("reopen commands failed, proxy: {}", e1
)),
867 (Ok(_
), Err(e2
)) => Err(format_err
!("reopen commands failed, api: {}", e2
)),
872 async
fn command_reopen_auth_logfiles() -> Result
<(), Error
> {
873 // only care about the most recent daemon instance for each, proxy & api, as other older ones
874 // should not respond to new requests anyway, but only finish their current one and then exit.
875 let sock
= proxmox_rest_server
::our_ctrl_sock();
876 let f1
= proxmox_rest_server
::send_raw_command(sock
, "{\"command\":\"api-auth-log-reopen\"}\n");
878 let pid
= proxmox_rest_server
::read_pid(pbs_buildcfg
::PROXMOX_BACKUP_API_PID_FN
)?
;
879 let sock
= proxmox_rest_server
::ctrl_sock_from_pid(pid
);
880 let f2
= proxmox_rest_server
::send_raw_command(sock
, "{\"command\":\"api-auth-log-reopen\"}\n");
882 match futures
::join
!(f1
, f2
) {
883 (Err(e1
), Err(e2
)) => Err(format_err
!("reopen commands failed, proxy: {}; api: {}", e1
, e2
)),
884 (Err(e1
), Ok(_
)) => Err(format_err
!("reopen commands failed, proxy: {}", e1
)),
885 (Ok(_
), Err(e2
)) => Err(format_err
!("reopen commands failed, api: {}", e2
)),
890 async
fn run_stat_generator() {
895 let save
= if count
>= 6 { count = 0; true }
else { false }
;
897 let delay_target
= Instant
::now() + Duration
::from_secs(10);
899 generate_host_stats(save
).await
;
901 tokio
::time
::sleep_until(tokio
::time
::Instant
::from_std(delay_target
)).await
;
907 fn rrd_update_gauge(name
: &str, value
: f64, save
: bool
) {
908 if let Err(err
) = RRD_CACHE
.update_value(name
, value
, DST
::Gauge
, save
) {
909 eprintln
!("rrd::update_value '{}' failed - {}", name
, err
);
913 fn rrd_update_derive(name
: &str, value
: f64, save
: bool
) {
914 if let Err(err
) = RRD_CACHE
.update_value(name
, value
, DST
::Derive
, save
) {
915 eprintln
!("rrd::update_value '{}' failed - {}", name
, err
);
919 async
fn generate_host_stats(save
: bool
) {
920 use proxmox
::sys
::linux
::procfs
::{
921 read_meminfo
, read_proc_stat
, read_proc_net_dev
, read_loadavg
};
923 pbs_runtime
::block_in_place(move || {
925 match read_proc_stat() {
927 rrd_update_gauge("host/cpu", stat
.cpu
, save
);
928 rrd_update_gauge("host/iowait", stat
.iowait_percent
, save
);
931 eprintln
!("read_proc_stat failed - {}", err
);
935 match read_meminfo() {
937 rrd_update_gauge("host/memtotal", meminfo
.memtotal
as f64, save
);
938 rrd_update_gauge("host/memused", meminfo
.memused
as f64, save
);
939 rrd_update_gauge("host/swaptotal", meminfo
.swaptotal
as f64, save
);
940 rrd_update_gauge("host/swapused", meminfo
.swapused
as f64, save
);
943 eprintln
!("read_meminfo failed - {}", err
);
947 match read_proc_net_dev() {
949 use pbs_config
::network
::is_physical_nic
;
953 if !is_physical_nic(&item
.device
) { continue; }
954 netin
+= item
.receive
;
957 rrd_update_derive("host/netin", netin
as f64, save
);
958 rrd_update_derive("host/netout", netout
as f64, save
);
961 eprintln
!("read_prox_net_dev failed - {}", err
);
965 match read_loadavg() {
967 rrd_update_gauge("host/loadavg", loadavg
.0 as f64, save
);
970 eprintln
!("read_loadavg failed - {}", err
);
974 let disk_manager
= DiskManage
::new();
976 gather_disk_stats(disk_manager
.clone(), Path
::new("/"), "host", save
);
978 match pbs_config
::datastore
::config() {
980 let datastore_list
: Vec
<DataStoreConfig
> =
981 config
.convert_to_typed_array("datastore").unwrap_or_default();
983 for config
in datastore_list
{
985 let rrd_prefix
= format
!("datastore/{}", config
.name
);
986 let path
= std
::path
::Path
::new(&config
.path
);
987 gather_disk_stats(disk_manager
.clone(), path
, &rrd_prefix
, save
);
991 eprintln
!("read datastore config failed - {}", err
);
998 fn check_schedule(worker_type
: &str, event_str
: &str, id
: &str) -> bool
{
999 let event
= match parse_calendar_event(event_str
) {
1002 eprintln
!("unable to parse schedule '{}' - {}", event_str
, err
);
1007 let last
= match jobstate
::last_run_time(worker_type
, &id
) {
1010 eprintln
!("could not get last run time of {} {}: {}", worker_type
, id
, err
);
1015 let next
= match compute_next_event(&event
, last
, false) {
1016 Ok(Some(next
)) => next
,
1017 Ok(None
) => return false,
1019 eprintln
!("compute_next_event for '{}' failed - {}", event_str
, err
);
1024 let now
= proxmox_time
::epoch_i64();
1028 fn gather_disk_stats(disk_manager
: Arc
<DiskManage
>, path
: &Path
, rrd_prefix
: &str, save
: bool
) {
1030 match proxmox_backup
::tools
::disks
::disk_usage(path
) {
1032 let rrd_key
= format
!("{}/total", rrd_prefix
);
1033 rrd_update_gauge(&rrd_key
, status
.total
as f64, save
);
1034 let rrd_key
= format
!("{}/used", rrd_prefix
);
1035 rrd_update_gauge(&rrd_key
, status
.used
as f64, save
);
1038 eprintln
!("read disk_usage on {:?} failed - {}", path
, err
);
1042 match disk_manager
.find_mounted_device(path
) {
1044 Ok(Some((fs_type
, device
, source
))) => {
1045 let mut device_stat
= None
;
1046 match fs_type
.as_str() {
1048 if let Some(source
) = source
{
1049 let pool
= get_pool_from_dataset(&source
).unwrap_or(&source
);
1050 match zfs_pool_stats(pool
) {
1051 Ok(stat
) => device_stat
= stat
,
1052 Err(err
) => eprintln
!("zfs_pool_stats({:?}) failed - {}", pool
, err
),
1057 if let Ok(disk
) = disk_manager
.clone().disk_by_dev_num(device
.into_dev_t()) {
1058 match disk
.read_stat() {
1059 Ok(stat
) => device_stat
= stat
,
1060 Err(err
) => eprintln
!("disk.read_stat {:?} failed - {}", path
, err
),
1065 if let Some(stat
) = device_stat
{
1066 let rrd_key
= format
!("{}/read_ios", rrd_prefix
);
1067 rrd_update_derive(&rrd_key
, stat
.read_ios
as f64, save
);
1068 let rrd_key
= format
!("{}/read_bytes", rrd_prefix
);
1069 rrd_update_derive(&rrd_key
, (stat
.read_sectors
*512) as f64, save
);
1071 let rrd_key
= format
!("{}/write_ios", rrd_prefix
);
1072 rrd_update_derive(&rrd_key
, stat
.write_ios
as f64, save
);
1073 let rrd_key
= format
!("{}/write_bytes", rrd_prefix
);
1074 rrd_update_derive(&rrd_key
, (stat
.write_sectors
*512) as f64, save
);
1076 let rrd_key
= format
!("{}/io_ticks", rrd_prefix
);
1077 rrd_update_derive(&rrd_key
, (stat
.io_ticks
as f64)/1000.0, save
);
1081 eprintln
!("find_mounted_device failed - {}", err
);