tyler/proxmox-backup
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

move worker_task.rs into proxmox-rest-server crate

Browse Source 
Also moved pbs-datastore/src/task.rs to pbs-tools, which now depends on 'log'.
This commit is contained in:
Dietmar Maurer
2021-09-23 10:09:19 +02:00
parent 81867f0539
commit b9700a9fe5
46 changed files with 176 additions and 183 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/acme/plugin.rs
View File

@ -13,7 +13,7 @@ use proxmox_acme_rs::{Authorization, Challenge};
use crate::acme::AcmeClient;
use crate::api2::types::AcmeDomain;
use crate::server::WorkerTask;
use proxmox_rest_server::WorkerTask;
use crate::config::acme::plugin::{DnsPlugin, PluginData};

4
src/api2/admin/datastore.rs
View File

@ -54,7 +54,7 @@ use pbs_tools::blocking::WrappedReaderStream;
use pbs_tools::stream::{AsyncReaderStream, AsyncChannelWriter};
use pbs_tools::json::{required_integer_param, required_string_param};
use pbs_config::CachedUserInfo;
use proxmox_rest_server::formatter;
use proxmox_rest_server::{WorkerTask, formatter};
use crate::api2::node::rrd::create_value_from_rrd;
use crate::backup::{
@ -62,7 +62,7 @@ use crate::backup::{
DataStore, LocalChunkReader,
};
use crate::server::{jobstate::Job, WorkerTask};
use crate::server::jobstate::Job;
const GROUP_NOTES_FILE_NAME: &str = "notes";

4
src/api2/backup/environment.rs
View File

@ -15,10 +15,10 @@ use pbs_datastore::backup_info::{BackupDir, BackupInfo};
use pbs_datastore::dynamic_index::DynamicIndexWriter;
use pbs_datastore::fixed_index::FixedIndexWriter;
use pbs_api_types::Authid;
use proxmox_rest_server::formatter::*;
use proxmox_rest_server::{WorkerTask, formatter::*};
use crate::backup::{verify_backup_dir_with_lock, DataStore};
use crate::server::WorkerTask;
use hyper::{Body, Response};
#[derive(Copy, Clone, Serialize)]

3
src/api2/backup/mod.rs
View File

@ -23,8 +23,9 @@ use pbs_datastore::PROXMOX_BACKUP_PROTOCOL_ID_V1;
use pbs_datastore::backup_info::{BackupDir, BackupGroup, BackupInfo};
use pbs_datastore::index::IndexFile;
use pbs_datastore::manifest::{archive_type, ArchiveType};
use proxmox_rest_server::WorkerTask;
use crate::server::{WorkerTask, H2Service};
use crate::server::H2Service;
use crate::backup::DataStore;
use pbs_config::CachedUserInfo;

2
src/api2/config/acme.rs
View File

@ -24,7 +24,7 @@ use crate::api2::types::{AcmeAccountName, AcmeChallengeSchema, KnownAcmeDirector
use crate::config::acme::plugin::{
self, DnsPlugin, DnsPluginCore, DnsPluginCoreUpdater, PLUGIN_ID_SCHEMA,
};
use crate::server::WorkerTask;
use proxmox_rest_server::WorkerTask;
pub(crate) const ROUTER: Router = Router::new()
.get(&list_subdirs_api_method!(SUBDIRS))

5
src/api2/config/datastore.rs
View File

@ -9,7 +9,6 @@ use proxmox::api::section_config::SectionConfigData;
use proxmox::api::schema::{ApiType, parse_property_string};
use pbs_datastore::chunk_store::ChunkStore;
use pbs_datastore::task::TaskState;
use pbs_config::BackupLockGuard;
use pbs_api_types::{
Authid, DatastoreNotify,
@ -17,6 +16,7 @@ use pbs_api_types::{
PRIV_DATASTORE_ALLOCATE, PRIV_DATASTORE_AUDIT, PRIV_DATASTORE_MODIFY,
DataStoreConfig, DataStoreConfigUpdater,
};
use pbs_tools::task::TaskState;
use crate::api2::config::sync::delete_sync_job;
use crate::api2::config::verify::delete_verification_job;
@ -26,8 +26,9 @@ use crate::api2::admin::{
verify::list_verification_jobs,
};
use pbs_config::CachedUserInfo;
use proxmox_rest_server::WorkerTask;
use crate::server::{jobstate, WorkerTask};
use crate::server::jobstate;
#[api(
input: {

2
src/api2/node/apt.rs
View File

@ -19,7 +19,7 @@ use pbs_api_types::{
};
use crate::config::node;
use crate::server::WorkerTask;
use proxmox_rest_server::WorkerTask;
use crate::tools::{
apt,
pbs_simple_http,

2
src/api2/node/certificates.rs
View File

@ -18,7 +18,7 @@ use pbs_tools::cert;
use crate::acme::AcmeClient;
use crate::api2::types::AcmeDomain;
use crate::config::node::NodeConfig;
use crate::server::WorkerTask;
use proxmox_rest_server::WorkerTask;
pub const ROUTER: Router = Router::new()
.get(&list_subdirs_api_method!(SUBDIRS))

2
src/api2/node/disks/directory.rs
View File

@ -17,7 +17,7 @@ use crate::tools::disks::{
};
use crate::tools::systemd::{self, types::*};
use crate::server::WorkerTask;
use proxmox_rest_server::WorkerTask;
const BASE_MOUNT_DIR: &str = "/mnt/datastore/";

2
src/api2/node/disks/mod.rs
View File

@ -15,7 +15,7 @@ use crate::tools::disks::{
DiskUsageInfo, DiskUsageType, DiskManage, SmartData,
get_disks, get_smart_data, get_disk_usage_info, inititialize_gpt_disk,
};
use crate::server::WorkerTask;
use proxmox_rest_server::WorkerTask;
pub mod directory;
pub mod zfs;

2
src/api2/node/disks/zfs.rs
View File

@ -19,7 +19,7 @@ use crate::tools::disks::{
DiskUsageType,
};
use crate::server::WorkerTask;
use proxmox_rest_server::WorkerTask;
#[api(

2
src/api2/node/mod.rs
View File

@ -24,7 +24,7 @@ use pbs_api_types::{Authid, NODE_SCHEMA, PRIV_SYS_CONSOLE};
use pbs_tools::auth::private_auth_key;
use pbs_tools::ticket::{self, Empty, Ticket};
use crate::server::WorkerTask;
use proxmox_rest_server::WorkerTask;
use crate::tools;
pub mod apt;

2
src/api2/node/network.rs
View File

@ -13,7 +13,7 @@ use pbs_api_types::{
};
use pbs_config::network::{self, NetworkConfig};
use crate::server::{WorkerTask};
use proxmox_rest_server::WorkerTask;
fn split_interface_list(list: &str) -> Result<Vec<String>, Error> {
let value = parse_property_string(&list, &NETWORK_INTERFACE_ARRAY_SCHEMA)?;

2
src/api2/node/services.rs
View File

@ -9,7 +9,7 @@ use proxmox::api::router::SubdirMap;
use pbs_api_types::{Authid, NODE_SCHEMA, SERVICE_ID_SCHEMA, PRIV_SYS_AUDIT, PRIV_SYS_MODIFY};
use crate::server::WorkerTask;
use proxmox_rest_server::WorkerTask;
static SERVICE_NAME_LIST: [&str; 7] = [
"proxmox-backup",

32
src/api2/node/tasks.rs
View File

@ -16,7 +16,8 @@ use pbs_api_types::{
};
use crate::api2::pull::check_pull_privs;
use crate::server::{self, upid_log_path, upid_read_status, TaskState, TaskListInfoIterator};
use proxmox_rest_server::{upid_log_path, upid_read_status, TaskState, TaskListInfoIterator};
use pbs_config::CachedUserInfo;
// matches respective job execution privileges
@ -125,6 +126,25 @@ pub fn tasktype(state: &TaskState) -> TaskStateType {
}
}
fn into_task_list_item(info: proxmox_rest_server::TaskListInfo) -> pbs_api_types::TaskListItem {
let (endtime, status) = info
.state
.map_or_else(|| (None, None), |a| (Some(a.endtime()), Some(a.to_string())));
pbs_api_types::TaskListItem {
upid: info.upid_str,
node: "localhost".to_string(),
pid: info.upid.pid as i64,
pstart: info.upid.pstart,
starttime: info.upid.starttime,
worker_type: info.upid.worker_type,
worker_id: info.upid.worker_id,
user: info.upid.auth_id,
endtime,
status,
}
}
#[api(
input: {
properties: {
@ -217,7 +237,7 @@ async fn get_task_status(
result["tokenid"] = Value::from(task_auth_id.tokenname().unwrap().as_str());
}
if crate::server::worker_is_active(&upid).await? {
if proxmox_rest_server::worker_is_active(&upid).await? {
result["status"] = Value::from("running");
} else {
let exitstatus = upid_read_status(&upid).unwrap_or(TaskState::Unknown { endtime: 0 });
@ -314,7 +334,7 @@ async fn read_task_log(
rpcenv["total"] = Value::from(count);
if test_status {
let active = crate::server::worker_is_active(&upid).await?;
let active = proxmox_rest_server::worker_is_active(&upid).await?;
rpcenv["active"] = Value::from(active);
}
@ -354,7 +374,7 @@ fn stop_task(
user_info.check_privs(&auth_id, &["system", "tasks"], PRIV_SYS_MODIFY, false)?;
}
server::abort_worker_async(upid);
proxmox_rest_server::abort_worker_async(upid);
Ok(Value::Null)
}
@ -502,7 +522,7 @@ pub fn list_tasks(
match (&info.state, &statusfilter) {
(Some(_), _) if running => continue,
(Some(crate::server::TaskState::OK { .. }), _) if errors => continue,
(Some(TaskState::OK { .. }), _) if errors => continue,
(Some(state), Some(filters)) => {
if !filters.contains(&tasktype(state)) {
continue;
@ -517,7 +537,7 @@ pub fn list_tasks(
continue;
}
result.push(info.into());
result.push(into_task_list_item(info));
if result.len() >= limit {
break;

3
src/api2/pull.rs
View File

@ -13,8 +13,9 @@ use pbs_api_types::{
DATASTORE_SCHEMA, REMOTE_ID_SCHEMA, REMOVE_VANISHED_BACKUPS_SCHEMA,
PRIV_DATASTORE_BACKUP, PRIV_DATASTORE_PRUNE, PRIV_REMOTE_READ,
};
use proxmox_rest_server::WorkerTask;
use crate::server::{WorkerTask, jobstate::Job, pull::pull_store};
use crate::server::{jobstate::Job, pull::pull_store};
use crate::backup::DataStore;
use pbs_config::CachedUserInfo;

2
src/api2/reader/environment.rs
View File

@ -10,7 +10,7 @@ use pbs_api_types::Authid;
use proxmox_rest_server::formatter::*;
use crate::backup::DataStore;
use crate::server::WorkerTask;
use proxmox_rest_server::WorkerTask;
//use proxmox::tools;

2
src/api2/reader/mod.rs
View File

@ -39,12 +39,12 @@ use pbs_datastore::backup_info::BackupDir;
use pbs_datastore::index::IndexFile;
use pbs_datastore::manifest::{archive_type, ArchiveType};
use pbs_config::CachedUserInfo;
use proxmox_rest_server::WorkerTask;
use crate::{
api2::helpers,
backup::DataStore,
server::{
WorkerTask,
H2Service,
},
};

6
src/api2/tape/backup.rs
View File

@ -20,10 +20,11 @@ use pbs_api_types::{
UPID_SCHEMA, JOB_ID_SCHEMA, PRIV_DATASTORE_READ, PRIV_TAPE_AUDIT, PRIV_TAPE_WRITE,
};
use pbs_datastore::{task_log, task_warn, StoreProgress};
use pbs_datastore::StoreProgress;
use pbs_datastore::backup_info::{BackupDir, BackupInfo};
use pbs_datastore::task::TaskState;
use pbs_tools::{task_log, task_warn, task::TaskState};
use pbs_config::CachedUserInfo;
use proxmox_rest_server::WorkerTask;
use crate::{
server::{
@ -36,7 +37,6 @@ use crate::{
},
},
backup::{DataStore, SnapshotReader},
server::WorkerTask,
tape::{
TAPE_STATUS_DIR,
Inventory,

4
src/api2/tape/drive.rs
View File

@ -28,7 +28,6 @@ use pbs_api_types::{
LtoDriveAndMediaStatus, Lp17VolumeStatistics,
};
use pbs_datastore::task_log;
use pbs_api_types::{PRIV_TAPE_AUDIT, PRIV_TAPE_READ, PRIV_TAPE_WRITE};
use pbs_config::CachedUserInfo;
use pbs_tape::{
@ -36,13 +35,14 @@ use pbs_tape::{
sg_tape::tape_alert_flags_critical,
linux_list_drives::{lto_tape_device_list, lookup_device_identification, open_lto_tape_device},
};
use pbs_tools::task_log;
use proxmox_rest_server::WorkerTask;
use crate::{
api2::tape::restore::{
fast_catalog_restore,
restore_media,
},
server::WorkerTask,
tape::{
TAPE_STATUS_DIR,
Inventory,

10
src/api2/tape/restore.rs
View File

@ -34,26 +34,24 @@ use pbs_api_types::{
UPID_SCHEMA, TAPE_RESTORE_SNAPSHOT_SCHEMA,
PRIV_DATASTORE_BACKUP, PRIV_DATASTORE_MODIFY, PRIV_TAPE_READ,
};
use pbs_datastore::{task_log, task_warn, DataBlob};
use pbs_datastore::DataBlob;
use pbs_datastore::backup_info::BackupDir;
use pbs_datastore::dynamic_index::DynamicIndexReader;
use pbs_datastore::fixed_index::FixedIndexReader;
use pbs_datastore::index::IndexFile;
use pbs_datastore::manifest::{archive_type, ArchiveType, BackupManifest, MANIFEST_BLOB_NAME};
use pbs_datastore::task::TaskState;
use pbs_config::CachedUserInfo;
use pbs_tape::{
TapeRead, BlockReadError, MediaContentHeader,
PROXMOX_BACKUP_CONTENT_HEADER_MAGIC_1_0,
};
use pbs_tools::{task_log, task_warn, task::TaskState};
use proxmox_rest_server::WorkerTask;
use crate::{
tools::ParallelHandler,
backup::DataStore,
server::{
lookup_user_email,
WorkerTask,
},
server::lookup_user_email,
tape::{
TAPE_STATUS_DIR,
MediaId,

3
src/backup/datastore.rs
View File

@ -12,7 +12,6 @@ use lazy_static::lazy_static;
use proxmox::tools::fs::{replace_file, file_read_optional_string, CreateOptions};
use pbs_api_types::{UPID, DataStoreConfig, Authid, GarbageCollectionStatus};
use pbs_datastore::{task_log, task_warn};
use pbs_datastore::DataBlob;
use pbs_datastore::backup_info::{BackupGroup, BackupDir};
use pbs_datastore::chunk_store::ChunkStore;
@ -24,10 +23,10 @@ use pbs_datastore::manifest::{
ArchiveType, BackupManifest,
archive_type,
};
use pbs_datastore::task::TaskState;
use pbs_tools::format::HumanByte;
use pbs_tools::fs::{lock_dir_noblock, DirLockGuard};
use pbs_tools::process_locker::ProcessLockSharedGuard;
use pbs_tools::{task_log, task_warn, task::TaskState};
use pbs_config::{open_backup_lockfile, BackupLockGuard};
use proxmox_rest_server::fail_on_shutdown;

4
src/backup/verify.rs
View File

@ -7,12 +7,12 @@ use std::time::Instant;
use anyhow::{bail, format_err, Error};
use pbs_api_types::{Authid, CryptMode, VerifyState, UPID, SnapshotVerifyState};
use pbs_datastore::{task_log, DataBlob, StoreProgress};
use pbs_datastore::{DataBlob, StoreProgress};
use pbs_datastore::backup_info::{BackupGroup, BackupDir, BackupInfo};
use pbs_datastore::index::IndexFile;
use pbs_datastore::manifest::{archive_type, ArchiveType, BackupManifest, FileInfo};
use pbs_datastore::task::TaskState;
use pbs_tools::fs::lock_dir_noblock_shared;
use pbs_tools::{task_log, task::TaskState};
use crate::{
backup::DataStore,

17
src/bin/proxmox-backup-api.rs
View File

@ -10,14 +10,9 @@ use proxmox::api::RpcEnvironmentType;
use proxmox::tools::fs::CreateOptions;
use pbs_tools::auth::private_auth_key;
use proxmox_rest_server::{ApiConfig, RestServer};
use proxmox_backup::server::{
self,
auth::default_api_auth,
};
use proxmox_rest_server::daemon;
use proxmox_rest_server::{daemon, ApiConfig, RestServer};
use proxmox_backup::server::auth::default_api_auth;
use proxmox_backup::auth_helpers::*;
use proxmox_backup::config;
@ -86,7 +81,7 @@ async fn run() -> Result<(), Error> {
)?;
let backup_user = pbs_config::backup_user()?;
let mut commando_sock = proxmox_rest_server::CommandoSocket::new(crate::server::our_ctrl_sock(), backup_user.gid);
let mut commando_sock = proxmox_rest_server::CommandoSocket::new(proxmox_rest_server::our_ctrl_sock(), backup_user.gid);
let dir_opts = CreateOptions::new().owner(backup_user.uid).group(backup_user.gid);
let file_opts = CreateOptions::new().owner(backup_user.uid).group(backup_user.gid);
@ -107,7 +102,7 @@ async fn run() -> Result<(), Error> {
let rest_server = RestServer::new(config);
proxmox_backup::server::init_worker_tasks(pbs_buildcfg::PROXMOX_BACKUP_LOG_DIR_M!().into(), file_opts.clone())?;
proxmox_rest_server::init_worker_tasks(pbs_buildcfg::PROXMOX_BACKUP_LOG_DIR_M!().into(), file_opts.clone())?;
// http server future:
let server = daemon::create_daemon(
@ -130,11 +125,11 @@ async fn run() -> Result<(), Error> {
"proxmox-backup.service",
);
server::write_pid(pbs_buildcfg::PROXMOX_BACKUP_API_PID_FN)?;
proxmox_rest_server::write_pid(pbs_buildcfg::PROXMOX_BACKUP_API_PID_FN)?;
daemon::systemd_notify(daemon::SystemdNotify::Ready)?;
let init_result: Result<(), Error> = try_block!({
server::register_task_control_commands(&mut commando_sock)?;
proxmox_rest_server::register_task_control_commands(&mut commando_sock)?;
commando_sock.spawn()?;
proxmox_rest_server::server_state_init()?;
Ok(())

3
src/bin/proxmox-backup-manager.rs
View File

@ -14,9 +14,10 @@ use pbs_api_types::{
IGNORE_VERIFIED_BACKUPS_SCHEMA, VERIFICATION_OUTDATED_AFTER_SCHEMA,
};
use proxmox_rest_server::wait_for_local_worker;
use proxmox_backup::config;
use proxmox_backup::api2;
use proxmox_backup::server::wait_for_local_worker;
mod proxmox_backup_manager;
use proxmox_backup_manager::*;

24
src/bin/proxmox-backup-proxy.rs
View File

@ -19,18 +19,16 @@ use proxmox::api::RpcEnvironmentType;
use proxmox::sys::linux::socket::set_tcp_keepalive;
use proxmox::tools::fs::CreateOptions;
use proxmox_rest_server::{ApiConfig, RestServer};
use proxmox_rest_server::{rotate_task_log_archive, ApiConfig, RestServer, WorkerTask};
use proxmox_backup::{
backup::DataStore,
server::{
auth::default_api_auth,
WorkerTask,
jobstate::{
self,
Job,
},
rotate_task_log_archive,
},
};
@ -188,7 +186,7 @@ async fn run() -> Result<(), Error> {
config.register_template("console", "/usr/share/pve-xtermjs/index.html.hbs")?;
let backup_user = pbs_config::backup_user()?;
let mut commando_sock = proxmox_rest_server::CommandoSocket::new(crate::server::our_ctrl_sock(), backup_user.gid);
let mut commando_sock = proxmox_rest_server::CommandoSocket::new(proxmox_rest_server::our_ctrl_sock(), backup_user.gid);
let dir_opts = CreateOptions::new().owner(backup_user.uid).group(backup_user.gid);
let file_opts = CreateOptions::new().owner(backup_user.uid).group(backup_user.gid);
@ -208,7 +206,7 @@ async fn run() -> Result<(), Error> {
)?;
let rest_server = RestServer::new(config);
proxmox_backup::server::init_worker_tasks(pbs_buildcfg::PROXMOX_BACKUP_LOG_DIR_M!().into(), file_opts.clone())?;
proxmox_rest_server::init_worker_tasks(pbs_buildcfg::PROXMOX_BACKUP_LOG_DIR_M!().into(), file_opts.clone())?;
//openssl req -x509 -newkey rsa:4096 -keyout /etc/proxmox-backup/proxy.key -out /etc/proxmox-backup/proxy.pem -nodes
@ -266,11 +264,11 @@ async fn run() -> Result<(), Error> {
"proxmox-backup-proxy.service",
);
server::write_pid(pbs_buildcfg::PROXMOX_BACKUP_PROXY_PID_FN)?;
proxmox_rest_server::write_pid(pbs_buildcfg::PROXMOX_BACKUP_PROXY_PID_FN)?;
daemon::systemd_notify(daemon::SystemdNotify::Ready)?;
let init_result: Result<(), Error> = try_block!({
server::register_task_control_commands(&mut commando_sock)?;
proxmox_rest_server::register_task_control_commands(&mut commando_sock)?;
commando_sock.spawn()?;
proxmox_rest_server::server_state_init()?;
Ok(())
@ -806,11 +804,11 @@ async fn schedule_task_log_rotate() {
async fn command_reopen_access_logfiles() -> Result<(), Error> {
// only care about the most recent daemon instance for each, proxy & api, as other older ones
// should not respond to new requests anyway, but only finish their current one and then exit.
let sock = crate::server::our_ctrl_sock();
let sock = proxmox_rest_server::our_ctrl_sock();
let f1 = proxmox_rest_server::send_command(sock, "{\"command\":\"api-access-log-reopen\"}\n");
let pid = crate::server::read_pid(pbs_buildcfg::PROXMOX_BACKUP_API_PID_FN)?;
let sock = crate::server::ctrl_sock_from_pid(pid);
let pid = proxmox_rest_server::read_pid(pbs_buildcfg::PROXMOX_BACKUP_API_PID_FN)?;
let sock = proxmox_rest_server::ctrl_sock_from_pid(pid);
let f2 = proxmox_rest_server::send_command(sock, "{\"command\":\"api-access-log-reopen\"}\n");
match futures::join!(f1, f2) {
@ -824,11 +822,11 @@ async fn command_reopen_access_logfiles() -> Result<(), Error> {
async fn command_reopen_auth_logfiles() -> Result<(), Error> {
// only care about the most recent daemon instance for each, proxy & api, as other older ones
// should not respond to new requests anyway, but only finish their current one and then exit.
let sock = crate::server::our_ctrl_sock();
let sock = proxmox_rest_server::our_ctrl_sock();
let f1 = proxmox_rest_server::send_command(sock, "{\"command\":\"api-auth-log-reopen\"}\n");
let pid = crate::server::read_pid(pbs_buildcfg::PROXMOX_BACKUP_API_PID_FN)?;
let sock = crate::server::ctrl_sock_from_pid(pid);
let pid = proxmox_rest_server::read_pid(pbs_buildcfg::PROXMOX_BACKUP_API_PID_FN)?;
let sock = proxmox_rest_server::ctrl_sock_from_pid(pid);
let f2 = proxmox_rest_server::send_command(sock, "{\"command\":\"api-auth-log-reopen\"}\n");
match futures::join!(f1, f2) {

2
src/bin/proxmox-daily-update.rs
View File

@ -11,7 +11,7 @@ async fn wait_for_local_worker(upid_str: &str) -> Result<(), Error> {
let sleep_duration = core::time::Duration::new(0, 100_000_000);
loop {
if !proxmox_backup::server::worker_is_active_local(&upid) {
if !proxmox_rest_server::worker_is_active_local(&upid) {
break;
}
tokio::time::sleep(sleep_duration).await;

4
src/bin/proxmox_backup_debug/api.rs
View File

@ -235,12 +235,12 @@ async fn handle_worker(upid_str: &str) -> Result<(), Error> {
let abort_future = async move {
while signal_stream.recv().await.is_some() {
println!("got shutdown request (SIGINT)");
proxmox_backup::server::abort_local_worker(upid.clone());
proxmox_rest_server::abort_local_worker(upid.clone());
}
Ok::<_, Error>(())
};
let result_future = proxmox_backup::server::wait_for_local_worker(upid_str);
let result_future = proxmox_rest_server::wait_for_local_worker(upid_str);
futures::select! {
result = result_future.fuse() => result?,

2
src/server/gc_job.rs
View File

@ -2,9 +2,9 @@ use std::sync::Arc;
use anyhow::Error;
use pbs_api_types::Authid;
use proxmox_rest_server::WorkerTask;
use crate::{
server::WorkerTask,
server::jobstate::Job,
backup::DataStore,
};

4
src/server/h2service.rs
View File

@ -11,11 +11,9 @@ use hyper::{Body, Request, Response, StatusCode};
use proxmox::api::{ApiResponseFuture, HttpError, Router, RpcEnvironment};
use proxmox::http_err;
use proxmox_rest_server::normalize_uri_path;
use proxmox_rest_server::{normalize_uri_path, WorkerTask};
use proxmox_rest_server::formatter::*;
use crate::server::WorkerTask;
/// Hyper Service implementation to handle stateful H2 connections.
///
/// We use this kind of service to handle backup protocol

8
src/server/jobstate.rs
View File

@ -14,7 +14,7 @@
//! an example usage would be
//! ```no_run
//! # use anyhow::{bail, Error};
//! # use proxmox_backup::server::TaskState;
//! # use proxmox_rest_server::TaskState;
//! # use proxmox_backup::server::jobstate::*;
//! # fn some_code() -> TaskState { TaskState::OK { endtime: 0 } }
//! # fn code() -> Result<(), Error> {
@ -50,11 +50,7 @@ use proxmox_systemd::time::{compute_next_event, parse_calendar_event};
use pbs_config::{open_backup_lockfile, BackupLockGuard};
use pbs_api_types::{UPID, JobScheduleStatus};
use crate::server::{
TaskState,
upid_read_status,
worker_is_active_local,
};
use proxmox_rest_server::{upid_read_status, worker_is_active_local, TaskState};
#[derive(Serialize, Deserialize)]
#[serde(rename_all = "kebab-case")]

48
src/server/mod.rs
View File

@ -4,51 +4,13 @@
//! services. We want async IO, so this is built on top of
//! tokio/hyper.
use anyhow::{format_err, Error};
use lazy_static::lazy_static;
use nix::unistd::Pid;
use anyhow::Error;
use serde_json::Value;
use proxmox::sys::linux::procfs::PidStat;
use proxmox::tools::fs::{create_path, CreateOptions};
use pbs_buildcfg;
lazy_static! {
static ref PID: i32 = unsafe { libc::getpid() };
static ref PSTART: u64 = PidStat::read_from_pid(Pid::from_raw(*PID)).unwrap().starttime;
}
pub fn pid() -> i32 {
*PID
}
pub fn pstart() -> u64 {
*PSTART
}
pub fn write_pid(pid_fn: &str) -> Result<(), Error> {
let pid_str = format!("{}\n", *PID);
proxmox::tools::fs::replace_file(pid_fn, pid_str.as_bytes(), CreateOptions::new())
}
pub fn read_pid(pid_fn: &str) -> Result<i32, Error> {
let pid = proxmox::tools::fs::file_get_contents(pid_fn)?;
let pid = std::str::from_utf8(&pid)?.trim();
pid.parse().map_err(|err| format_err!("could not parse pid - {}", err))
}
pub fn ctrl_sock_from_pid(pid: i32) -> String {
format!("\0{}/control-{}.sock", pbs_buildcfg::PROXMOX_BACKUP_RUN_DIR, pid)
}
pub fn our_ctrl_sock() -> String {
ctrl_sock_from_pid(*PID)
}
mod worker_task;
pub use worker_task::*;
mod h2service;
pub use h2service::*;
@ -76,16 +38,16 @@ pub mod auth;
pub mod pull;
pub(crate) async fn reload_proxy_certificate() -> Result<(), Error> {
let proxy_pid = crate::server::read_pid(pbs_buildcfg::PROXMOX_BACKUP_PROXY_PID_FN)?;
let sock = crate::server::ctrl_sock_from_pid(proxy_pid);
let proxy_pid = proxmox_rest_server::read_pid(pbs_buildcfg::PROXMOX_BACKUP_PROXY_PID_FN)?;
let sock = proxmox_rest_server::ctrl_sock_from_pid(proxy_pid);
let _: Value = proxmox_rest_server::send_raw_command(sock, "{\"command\":\"reload-certificate\"}\n")
.await?;
Ok(())
}
pub(crate) async fn notify_datastore_removed() -> Result<(), Error> {
let proxy_pid = crate::server::read_pid(pbs_buildcfg::PROXMOX_BACKUP_PROXY_PID_FN)?;
let sock = crate::server::ctrl_sock_from_pid(proxy_pid);
let proxy_pid = proxmox_rest_server::read_pid(pbs_buildcfg::PROXMOX_BACKUP_PROXY_PID_FN)?;
let sock = proxmox_rest_server::ctrl_sock_from_pid(proxy_pid);
let _: Value = proxmox_rest_server::send_raw_command(sock, "{\"command\":\"datastore-removed\"}\n")
.await?;
Ok(())

6
src/server/prune_job.rs
View File

@ -2,17 +2,17 @@ use std::sync::Arc;
use anyhow::Error;
use pbs_datastore::{task_log, task_warn};
use pbs_datastore::backup_info::BackupInfo;
use pbs_datastore::prune::compute_prune_info;
use pbs_api_types::{Authid, PRIV_DATASTORE_MODIFY, PruneOptions};
use pbs_config::CachedUserInfo;
use pbs_tools::{task_log, task_warn};
use proxmox_rest_server::WorkerTask;
use crate::{
backup::DataStore,
server::jobstate::Job,
server::WorkerTask,
};
};
pub fn prune_datastore(
worker: Arc<WorkerTask>,

5
src/server/pull.rs
View File

@ -13,7 +13,7 @@ use serde_json::json;
use proxmox::api::error::{HttpError, StatusCode};
use pbs_api_types::{Authid, SnapshotListItem, GroupListItem};
use pbs_datastore::{task_log, BackupInfo, BackupDir, BackupGroup, StoreProgress};
use pbs_datastore::{BackupInfo, BackupDir, BackupGroup, StoreProgress};
use pbs_datastore::data_blob::DataBlob;
use pbs_datastore::dynamic_index::DynamicIndexReader;
use pbs_datastore::fixed_index::FixedIndexReader;
@ -22,11 +22,12 @@ use pbs_datastore::manifest::{
CLIENT_LOG_BLOB_NAME, MANIFEST_BLOB_NAME, ArchiveType, BackupManifest, FileInfo, archive_type
};
use pbs_tools::sha::sha256;
use pbs_tools::task_log;
use pbs_client::{BackupReader, BackupRepository, HttpClient, HttpClientOptions, RemoteChunkReader};
use proxmox_rest_server::WorkerTask;
use crate::{
backup::DataStore,
server::WorkerTask,
tools::ParallelHandler,
};

4
src/server/verify_job.rs
View File

@ -1,10 +1,10 @@
use anyhow::{format_err, Error};
use pbs_datastore::task_log;
use pbs_tools::task_log;
use pbs_api_types::{Authid, VerificationJobConfig};
use proxmox_rest_server::WorkerTask;
use crate::{
server::WorkerTask,
server::jobstate::Job,
backup::{
DataStore,

923
src/server/worker_task.rs
View File

@ -1,923 +0,0 @@
use std::collections::{HashMap, VecDeque};
use std::fs::File;
use std::path::PathBuf;
use std::io::{Read, Write, BufRead, BufReader};
use std::panic::UnwindSafe;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use anyhow::{bail, format_err, Error};
use futures::*;
use lazy_static::lazy_static;
use serde_json::{json, Value};
use serde::{Serialize, Deserialize};
use tokio::sync::oneshot;
use nix::fcntl::OFlag;
use once_cell::sync::OnceCell;
use proxmox::sys::linux::procfs;
use proxmox::try_block;
use proxmox::tools::fs::{create_path, replace_file, atomic_open_or_create_file, CreateOptions};
use proxmox::api::upid::UPID;
use pbs_tools::logrotate::{LogRotate, LogRotateFiles};
use proxmox_rest_server::{CommandoSocket, FileLogger, FileLogOptions};
struct TaskListLockGuard(File);
struct WorkerTaskSetup {
file_opts: CreateOptions,
taskdir: PathBuf,
task_lock_fn: PathBuf,
active_tasks_fn: PathBuf,
task_index_fn: PathBuf,
task_archive_fn: PathBuf,
}
static WORKER_TASK_SETUP: OnceCell<WorkerTaskSetup> = OnceCell::new();
fn worker_task_setup() -> Result<&'static WorkerTaskSetup, Error> {
WORKER_TASK_SETUP.get()
.ok_or_else(|| format_err!("WorkerTask library is not initialized"))
}
impl WorkerTaskSetup {
fn new(basedir: PathBuf, file_opts: CreateOptions) -> Self {
let mut taskdir = basedir.clone();
taskdir.push("tasks");
let mut task_lock_fn = taskdir.clone();
task_lock_fn.push(".active.lock");
let mut active_tasks_fn = taskdir.clone();
active_tasks_fn.push("active");
let mut task_index_fn = taskdir.clone();
task_index_fn.push("index");
let mut task_archive_fn = taskdir.clone();
task_archive_fn.push("archive");
Self {
file_opts,
taskdir,
task_lock_fn,
active_tasks_fn,
task_index_fn,
task_archive_fn,
}
}
fn lock_task_list_files(&self, exclusive: bool) -> Result<TaskListLockGuard, Error> {
let options = self.file_opts.clone()
.perm(nix::sys::stat::Mode::from_bits_truncate(0o660));
let timeout = std::time::Duration::new(10, 0);
let file = proxmox::tools::fs::open_file_locked(
&self.task_lock_fn,
timeout,
exclusive,
options,
)?;
Ok(TaskListLockGuard(file))
}
fn log_path(&self, upid: &UPID) -> std::path::PathBuf {
let mut path = self.taskdir.clone();
path.push(format!("{:02X}", upid.pstart % 256));
path.push(upid.to_string());
path
}
// atomically read/update the task list, update status of finished tasks
// new_upid is added to the list when specified.
fn update_active_workers(&self, new_upid: Option<&UPID>) -> Result<(), Error> {
let lock = self.lock_task_list_files(true)?;
// TODO remove with 1.x
let mut finish_list: Vec<TaskListInfo> = read_task_file_from_path(&self.task_index_fn)?;
let had_index_file = !finish_list.is_empty();
// We use filter_map because one negative case wants to *move* the data into `finish_list`,
// clippy doesn't quite catch this!
#[allow(clippy::unnecessary_filter_map)]
let mut active_list: Vec<TaskListInfo> = read_task_file_from_path(&self.active_tasks_fn)?
.into_iter()
.filter_map(|info| {
if info.state.is_some() {
// this can happen when the active file still includes finished tasks
finish_list.push(info);
return None;
}
if !worker_is_active_local(&info.upid) {
// println!("Detected stopped task '{}'", &info.upid_str);
let now = proxmox::tools::time::epoch_i64();
let status = upid_read_status(&info.upid).unwrap_or(TaskState::Unknown { endtime: now });
finish_list.push(TaskListInfo {
upid: info.upid,
upid_str: info.upid_str,
state: Some(status)
});
return None;
}
Some(info)
}).collect();
if let Some(upid) = new_upid {
active_list.push(TaskListInfo { upid: upid.clone(), upid_str: upid.to_string(), state: None });
}
let active_raw = render_task_list(&active_list);
let options = self.file_opts.clone()
.perm(nix::sys::stat::Mode::from_bits_truncate(0o660));
replace_file(
&self.active_tasks_fn,
active_raw.as_bytes(),
options,
)?;
finish_list.sort_unstable_by(|a, b| {
match (&a.state, &b.state) {
(Some(s1), Some(s2)) => s1.cmp(&s2),
(Some(_), None) => std::cmp::Ordering::Less,
(None, Some(_)) => std::cmp::Ordering::Greater,
_ => a.upid.starttime.cmp(&b.upid.starttime),
}
});
if !finish_list.is_empty() {
let options = self.file_opts.clone()
.perm(nix::sys::stat::Mode::from_bits_truncate(0o660));
let mut writer = atomic_open_or_create_file(
&self.task_archive_fn,
OFlag::O_APPEND | OFlag::O_RDWR,
&[],
options,
)?;
for info in &finish_list {
writer.write_all(render_task_line(&info).as_bytes())?;
}
}
// TODO Remove with 1.x
// for compatibility, if we had an INDEX file, we do not need it anymore
if had_index_file {
let _ = nix::unistd::unlink(&self.task_index_fn);
}
drop(lock);
Ok(())
}
// Create task log directory with correct permissions
fn create_task_log_dirs(&self) -> Result<(), Error> {
try_block!({
let dir_opts = self.file_opts.clone()
.perm(nix::sys::stat::Mode::from_bits_truncate(0o755));
create_path(&self.taskdir, Some(dir_opts.clone()), Some(dir_opts.clone()))?;
// fixme:??? create_path(pbs_buildcfg::PROXMOX_BACKUP_RUN_DIR, None, Some(opts))?;
Ok(())
}).map_err(|err: Error| format_err!("unable to create task log dir - {}", err))
}
}
/// Initialize the WorkerTask library
pub fn init_worker_tasks(basedir: PathBuf, file_opts: CreateOptions) -> Result<(), Error> {
let setup = WorkerTaskSetup::new(basedir, file_opts);
setup.create_task_log_dirs()?;
WORKER_TASK_SETUP.set(setup)
.map_err(|_| format_err!("init_worker_tasks failed - already initialized"))
}
/// checks if the Task Archive is bigger that 'size_threshold' bytes, and
/// rotates it if it is
pub fn rotate_task_log_archive(size_threshold: u64, compress: bool, max_files: Option<usize>) -> Result<bool, Error> {
let setup = worker_task_setup()?;
let _lock = setup.lock_task_list_files(true)?;
let mut logrotate = LogRotate::new(&setup.task_archive_fn, compress)
.ok_or_else(|| format_err!("could not get archive file names"))?;
logrotate.rotate(size_threshold, None, max_files)
}
/// Path to the worker log file
pub fn upid_log_path(upid: &UPID) -> Result<std::path::PathBuf, Error> {
let setup = worker_task_setup()?;
Ok(setup.log_path(upid))
}
/// Read endtime (time of last log line) and exitstatus from task log file
/// If there is not a single line with at valid datetime, we assume the
/// starttime to be the endtime
pub fn upid_read_status(upid: &UPID) -> Result<TaskState, Error> {
let setup = worker_task_setup()?;
let mut status = TaskState::Unknown { endtime: upid.starttime };
let path = setup.log_path(upid);
let mut file = File::open(path)?;
/// speedup - only read tail
use std::io::Seek;
use std::io::SeekFrom;
let _ = file.seek(SeekFrom::End(-8192)); // ignore errors
let mut data = Vec::with_capacity(8192);
file.read_to_end(&mut data)?;
// strip newlines at the end of the task logs
while data.last() == Some(&b'\n') {
data.pop();
}
let last_line = match data.iter().rposition(|c| *c == b'\n') {
Some(start) if data.len() > (start+1) => &data[start+1..],
Some(_) => &data, // should not happen, since we removed all trailing newlines
None => &data,
};
let last_line = std::str::from_utf8(last_line)
.map_err(|err| format_err!("upid_read_status: utf8 parse failed: {}", err))?;
let mut iter = last_line.splitn(2, ": ");
if let Some(time_str) = iter.next() {
if let Ok(endtime) = proxmox::tools::time::parse_rfc3339(time_str) {
// set the endtime even if we cannot parse the state
status = TaskState::Unknown { endtime };
if let Some(rest) = iter.next().and_then(|rest| rest.strip_prefix("TASK ")) {
if let Ok(state) = TaskState::from_endtime_and_message(endtime, rest) {
status = state;
}
}
}
}
Ok(status)
}
lazy_static! {
static ref WORKER_TASK_LIST: Mutex<HashMap<usize, Arc<WorkerTask>>> = Mutex::new(HashMap::new());
}
/// checks if the task UPID refers to a worker from this process
fn is_local_worker(upid: &UPID) -> bool {
upid.pid == crate::server::pid() && upid.pstart == crate::server::pstart()
}
/// Test if the task is still running
pub async fn worker_is_active(upid: &UPID) -> Result<bool, Error> {
if is_local_worker(upid) {
return Ok(WORKER_TASK_LIST.lock().unwrap().contains_key(&upid.task_id));
}
if procfs::check_process_running_pstart(upid.pid, upid.pstart).is_none() {
return Ok(false);
}
let sock = crate::server::ctrl_sock_from_pid(upid.pid);
let cmd = json!({
"command": "worker-task-status",
"args": {
"upid": upid.to_string(),
},
});
let status = proxmox_rest_server::send_command(sock, &cmd).await?;
if let Some(active) = status.as_bool() {
Ok(active)
} else {
bail!("got unexpected result {:?} (expected bool)", status);
}
}
/// Test if the task is still running (fast but inaccurate implementation)
///
/// If the task is spawned from a different process, we simply return if
/// that process is still running. This information is good enough to detect
/// stale tasks...
pub fn worker_is_active_local(upid: &UPID) -> bool {
if is_local_worker(upid) {
WORKER_TASK_LIST.lock().unwrap().contains_key(&upid.task_id)
} else {
procfs::check_process_running_pstart(upid.pid, upid.pstart).is_some()
}
}
pub fn register_task_control_commands(
commando_sock: &mut CommandoSocket,
) -> Result<(), Error> {
fn get_upid(args: Option<&Value>) -> Result<UPID, Error> {
let args = if let Some(args) = args { args } else { bail!("missing args") };
let upid = match args.get("upid") {
Some(Value::String(upid)) => upid.parse::<UPID>()?,
None => bail!("no upid in args"),
_ => bail!("unable to parse upid"),
};
if !is_local_worker(&upid) {
bail!("upid does not belong to this process");
}
Ok(upid)
}
commando_sock.register_command("worker-task-abort".into(), move |args| {
let upid = get_upid(args)?;
abort_local_worker(upid);
Ok(Value::Null)
})?;
commando_sock.register_command("worker-task-status".into(), move |args| {
let upid = get_upid(args)?;
let active = WORKER_TASK_LIST.lock().unwrap().contains_key(&upid.task_id);
Ok(active.into())
})?;
Ok(())
}
pub fn abort_worker_async(upid: UPID) {
tokio::spawn(async move {
if let Err(err) = abort_worker(upid).await {
eprintln!("abort worker failed - {}", err);
}
});
}
pub async fn abort_worker(upid: UPID) -> Result<(), Error> {
let sock = crate::server::ctrl_sock_from_pid(upid.pid);
let cmd = json!({
"command": "worker-task-abort",
"args": {
"upid": upid.to_string(),
},
});
proxmox_rest_server::send_command(sock, &cmd).map_ok(|_| ()).await
}
fn parse_worker_status_line(line: &str) -> Result<(String, UPID, Option<TaskState>), Error> {
let data = line.splitn(3, ' ').collect::<Vec<&str>>();
let len = data.len();
match len {
1 => Ok((data[0].to_owned(), data[0].parse::<UPID>()?, None)),
3 => {
let endtime = i64::from_str_radix(data[1], 16)?;
let state = TaskState::from_endtime_and_message(endtime, data[2])?;
Ok((data[0].to_owned(), data[0].parse::<UPID>()?, Some(state)))
}
_ => bail!("wrong number of components"),
}
}
/// Task State
#[derive(Debug, PartialEq, Eq, Serialize, Deserialize)]
pub enum TaskState {
/// The Task ended with an undefined state
Unknown { endtime: i64 },
/// The Task ended and there were no errors or warnings
OK { endtime: i64 },
/// The Task had 'count' amount of warnings and no errors
Warning { count: u64, endtime: i64 },
/// The Task ended with the error described in 'message'
Error { message: String, endtime: i64 },
}
impl TaskState {
pub fn endtime(&self) -> i64 {
match *self {
TaskState::Unknown { endtime } => endtime,
TaskState::OK { endtime } => endtime,
TaskState::Warning { endtime, .. } => endtime,
TaskState::Error { endtime, .. } => endtime,
}
}
fn result_text(&self) -> String {
match self {
TaskState::Error { message, .. } => format!("TASK ERROR: {}", message),
other => format!("TASK {}", other),
}
}
fn from_endtime_and_message(endtime: i64, s: &str) -> Result<Self, Error> {
if s == "unknown" {
Ok(TaskState::Unknown { endtime })
} else if s == "OK" {
Ok(TaskState::OK { endtime })
} else if let Some(warnings) = s.strip_prefix("WARNINGS: ") {
let count: u64 = warnings.parse()?;
Ok(TaskState::Warning{ count, endtime })
} else if !s.is_empty() {
let message = if let Some(err) = s.strip_prefix("ERROR: ") { err } else { s }.to_string();
Ok(TaskState::Error{ message, endtime })
} else {
bail!("unable to parse Task Status '{}'", s);
}
}
}
impl std::cmp::PartialOrd for TaskState {
fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
Some(self.endtime().cmp(&other.endtime()))
}
}
impl std::cmp::Ord for TaskState {
fn cmp(&self, other: &Self) -> std::cmp::Ordering {
self.endtime().cmp(&other.endtime())
}
}
impl std::fmt::Display for TaskState {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
TaskState::Unknown { .. } => write!(f, "unknown"),
TaskState::OK { .. }=> write!(f, "OK"),
TaskState::Warning { count, .. } => write!(f, "WARNINGS: {}", count),
TaskState::Error { message, .. } => write!(f, "{}", message),
}
}
}
/// Task details including parsed UPID
///
/// If there is no `state`, the task is still running.
#[derive(Debug)]
pub struct TaskListInfo {
/// The parsed UPID
pub upid: UPID,
/// UPID string representation
pub upid_str: String,
/// Task `(endtime, status)` if already finished
pub state: Option<TaskState>, // endtime, status
}
impl Into<pbs_api_types::TaskListItem> for TaskListInfo {
fn into(self) -> pbs_api_types::TaskListItem {
let (endtime, status) = self
.state
.map_or_else(|| (None, None), |a| (Some(a.endtime()), Some(a.to_string())));
pbs_api_types::TaskListItem {
upid: self.upid_str,
node: "localhost".to_string(),
pid: self.upid.pid as i64,
pstart: self.upid.pstart,
starttime: self.upid.starttime,
worker_type: self.upid.worker_type,
worker_id: self.upid.worker_id,
user: self.upid.auth_id,
endtime,
status,
}
}
}
fn render_task_line(info: &TaskListInfo) -> String {
let mut raw = String::new();
if let Some(status) = &info.state {
raw.push_str(&format!("{} {:08X} {}\n", info.upid_str, status.endtime(), status));
} else {
raw.push_str(&info.upid_str);
raw.push('\n');
}
raw
}
fn render_task_list(list: &[TaskListInfo]) -> String {
let mut raw = String::new();
for info in list {
raw.push_str(&render_task_line(&info));
}
raw
}
// note this is not locked, caller has to make sure it is
// this will skip (and log) lines that are not valid status lines
fn read_task_file<R: Read>(reader: R) -> Result<Vec<TaskListInfo>, Error>
{
let reader = BufReader::new(reader);
let mut list = Vec::new();
for line in reader.lines() {
let line = line?;
match parse_worker_status_line(&line) {
Ok((upid_str, upid, state)) => list.push(TaskListInfo {
upid_str,
upid,
state
}),
Err(err) => {
eprintln!("unable to parse worker status '{}' - {}", line, err);
continue;
}
};
}
Ok(list)
}
// note this is not locked, caller has to make sure it is
fn read_task_file_from_path<P>(path: P) -> Result<Vec<TaskListInfo>, Error>
where
P: AsRef<std::path::Path> + std::fmt::Debug,
{
let file = match File::open(&path) {
Ok(f) => f,
Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
Err(err) => bail!("unable to open task list {:?} - {}", path, err),
};
read_task_file(file)
}
pub struct TaskListInfoIterator {
list: VecDeque<TaskListInfo>,
end: bool,
archive: Option<LogRotateFiles>,
lock: Option<TaskListLockGuard>,
}
impl TaskListInfoIterator {
pub fn new(active_only: bool) -> Result<Self, Error> {
let setup = worker_task_setup()?;
let (read_lock, active_list) = {
let lock = setup.lock_task_list_files(false)?;
let active_list = read_task_file_from_path(&setup.active_tasks_fn)?;
let needs_update = active_list
.iter()
.any(|info| info.state.is_some() || !worker_is_active_local(&info.upid));
// TODO remove with 1.x
let index_exists = setup.task_index_fn.is_file();
if needs_update || index_exists {
drop(lock);
setup.update_active_workers(None)?;
let lock = setup.lock_task_list_files(false)?;
let active_list = read_task_file_from_path(&setup.active_tasks_fn)?;
(lock, active_list)
} else {
(lock, active_list)
}
};
let archive = if active_only {
None
} else {
let logrotate = LogRotate::new(&setup.task_archive_fn, true)
.ok_or_else(|| format_err!("could not get archive file names"))?;
Some(logrotate.files())
};
let lock = if active_only { None } else { Some(read_lock) };
Ok(Self {
list: active_list.into(),
end: active_only,
archive,
lock,
})
}
}
impl Iterator for TaskListInfoIterator {
type Item = Result<TaskListInfo, Error>;
fn next(&mut self) -> Option<Self::Item> {
loop {
if let Some(element) = self.list.pop_back() {
return Some(Ok(element));
} else if self.end {
return None;
} else {
if let Some(mut archive) = self.archive.take() {
if let Some(file) = archive.next() {
let list = match read_task_file(file) {
Ok(list) => list,
Err(err) => return Some(Err(err)),
};
self.list.append(&mut list.into());
self.archive = Some(archive);
continue;
}
}
self.end = true;
self.lock.take();
}
}
}
}
/// Launch long running worker tasks.
///
/// A worker task can either be a whole thread, or a simply tokio
/// task/future. Each task can `log()` messages, which are stored
/// persistently to files. Task should poll the `abort_requested`
/// flag, and stop execution when requested.
pub struct WorkerTask {
setup: &'static WorkerTaskSetup,
upid: UPID,
data: Mutex<WorkerTaskData>,
abort_requested: AtomicBool,
}
impl std::fmt::Display for WorkerTask {
fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
self.upid.fmt(f)
}
}
struct WorkerTaskData {
logger: FileLogger,
progress: f64, // 0..1
warn_count: u64,
pub abort_listeners: Vec<oneshot::Sender<()>>,
}
impl WorkerTask {
pub fn new(
worker_type: &str,
worker_id: Option<String>,
auth_id: String,
to_stdout: bool,
) -> Result<Arc<Self>, Error> {
let setup = worker_task_setup()?;
let upid = UPID::new(worker_type, worker_id, auth_id)?;
let task_id = upid.task_id;
let mut path = setup.taskdir.clone();
path.push(format!("{:02X}", upid.pstart & 255));
let dir_opts = setup.file_opts.clone()
.perm(nix::sys::stat::Mode::from_bits_truncate(0o755));
create_path(&path, None, Some(dir_opts))?;
path.push(upid.to_string());
let logger_options = FileLogOptions {
to_stdout,
exclusive: true,
prefix_time: true,
read: true,
file_opts: setup.file_opts.clone(),
..Default::default()
};
let logger = FileLogger::new(&path, logger_options)?;
let worker = Arc::new(Self {
setup,
upid: upid.clone(),
abort_requested: AtomicBool::new(false),
data: Mutex::new(WorkerTaskData {
logger,
progress: 0.0,
warn_count: 0,
abort_listeners: vec![],
}),
});
// scope to drop the lock again after inserting
{
let mut hash = WORKER_TASK_LIST.lock().unwrap();
hash.insert(task_id, worker.clone());
proxmox_rest_server::set_worker_count(hash.len());
}
setup.update_active_workers(Some(&upid))?;
Ok(worker)
}
/// Spawn a new tokio task/future.
pub fn spawn<F, T>(
worker_type: &str,
worker_id: Option<String>,
auth_id: String,
to_stdout: bool,
f: F,
) -> Result<String, Error>
where F: Send + 'static + FnOnce(Arc<WorkerTask>) -> T,
T: Send + 'static + Future<Output = Result<(), Error>>,
{
let worker = WorkerTask::new(worker_type, worker_id, auth_id, to_stdout)?;
let upid_str = worker.upid.to_string();
let f = f(worker.clone());
tokio::spawn(async move {
let result = f.await;
worker.log_result(&result);
});
Ok(upid_str)
}
/// Create a new worker thread.
pub fn new_thread<F>(
worker_type: &str,
worker_id: Option<String>,
auth_id: String,
to_stdout: bool,
f: F,
) -> Result<String, Error>
where F: Send + UnwindSafe + 'static + FnOnce(Arc<WorkerTask>) -> Result<(), Error>
{
let worker = WorkerTask::new(worker_type, worker_id, auth_id, to_stdout)?;
let upid_str = worker.upid.to_string();
let _child = std::thread::Builder::new().name(upid_str.clone()).spawn(move || {
let worker1 = worker.clone();
let result = match std::panic::catch_unwind(move || f(worker1)) {
Ok(r) => r,
Err(panic) => {
match panic.downcast::<&str>() {
Ok(panic_msg) => {
Err(format_err!("worker panicked: {}", panic_msg))
}
Err(_) => {
Err(format_err!("worker panicked: unknown type."))
}
}
}
};
worker.log_result(&result);
});
Ok(upid_str)
}
/// create state from self and a result
pub fn create_state(&self, result: &Result<(), Error>) -> TaskState {
let warn_count = self.data.lock().unwrap().warn_count;
let endtime = proxmox::tools::time::epoch_i64();
if let Err(err) = result {
TaskState::Error { message: err.to_string(), endtime }
} else if warn_count > 0 {
TaskState::Warning { count: warn_count, endtime }
} else {
TaskState::OK { endtime }
}
}
/// Log task result, remove task from running list
pub fn log_result(&self, result: &Result<(), Error>) {
let state = self.create_state(result);
self.log(state.result_text());
WORKER_TASK_LIST.lock().unwrap().remove(&self.upid.task_id);
let _ = self.setup.update_active_workers(None);
proxmox_rest_server::set_worker_count(WORKER_TASK_LIST.lock().unwrap().len());
}
/// Log a message.
pub fn log<S: AsRef<str>>(&self, msg: S) {
let mut data = self.data.lock().unwrap();
data.logger.log(msg);
}
/// Log a message as warning.
pub fn warn<S: AsRef<str>>(&self, msg: S) {
let mut data = self.data.lock().unwrap();
data.logger.log(format!("WARN: {}", msg.as_ref()));
data.warn_count += 1;
}
/// Set progress indicator
pub fn progress(&self, progress: f64) {
if progress >= 0.0 && progress <= 1.0 {
let mut data = self.data.lock().unwrap();
data.progress = progress;
} else {
// fixme: log!("task '{}': ignoring strange value for progress '{}'", self.upid, progress);
}
}
/// Request abort
pub fn request_abort(&self) {
eprintln!("set abort flag for worker {}", self.upid);
let prev_abort = self.abort_requested.swap(true, Ordering::SeqCst);
if !prev_abort { // log abort one time
self.log(format!("received abort request ..."));
}
// noitify listeners
let mut data = self.data.lock().unwrap();
loop {
match data.abort_listeners.pop() {
None => { break; },
Some(ch) => {
let _ = ch.send(()); // ignore errors here
},
}
}
}
/// Test if abort was requested.
pub fn abort_requested(&self) -> bool {
self.abort_requested.load(Ordering::SeqCst)
}
/// Fail if abort was requested.
pub fn fail_on_abort(&self) -> Result<(), Error> {
if self.abort_requested() {
bail!("abort requested - aborting task");
}
Ok(())
}
/// Get a future which resolves on task abort
pub fn abort_future(&self) -> oneshot::Receiver<()> {
let (tx, rx) = oneshot::channel::<()>();
let mut data = self.data.lock().unwrap();
if self.abort_requested() {
let _ = tx.send(());
} else {
data.abort_listeners.push(tx);
}
rx
}
pub fn upid(&self) -> &UPID {
&self.upid
}
}
impl pbs_datastore::task::TaskState for WorkerTask {
fn check_abort(&self) -> Result<(), Error> {
self.fail_on_abort()
}
fn log(&self, level: log::Level, message: &std::fmt::Arguments) {
match level {
log::Level::Error => self.warn(&message.to_string()),
log::Level::Warn => self.warn(&message.to_string()),
log::Level::Info => self.log(&message.to_string()),
log::Level::Debug => self.log(&format!("DEBUG: {}", message)),
log::Level::Trace => self.log(&format!("TRACE: {}", message)),
}
}
}
/// Wait for a locally spanned worker task
///
/// Note: local workers should print logs to stdout, so there is no
/// need to fetch/display logs. We just wait for the worker to finish.
pub async fn wait_for_local_worker(upid_str: &str) -> Result<(), Error> {
let upid: UPID = upid_str.parse()?;
let sleep_duration = core::time::Duration::new(0, 100_000_000);
loop {
if worker_is_active_local(&upid) {
tokio::time::sleep(sleep_duration).await;
} else {
break;
}
}
Ok(())
}
/// Request abort of a local worker (if existing and running)
pub fn abort_local_worker(upid: UPID) {
if let Some(ref worker) = WORKER_TASK_LIST.lock().unwrap().get(&upid.task_id) {
worker.request_abort();
}
}

9
src/tape/drive/mod.rs
View File

@ -30,19 +30,16 @@ use proxmox::{
use pbs_api_types::{VirtualTapeDrive, LtoTapeDrive, Fingerprint};
use pbs_config::key_config::KeyConfig;
use pbs_datastore::task::TaskState;
use pbs_datastore::task_log;
use pbs_tools::{task_log, task::TaskState};
use pbs_tape::{
TapeWrite, TapeRead, BlockReadError, MediaContentHeader,
sg_tape::TapeAlertFlags,
};
use proxmox_rest_server::WorkerTask;
use crate::{
server::{
send_load_media_email,
WorkerTask,
},
server::send_load_media_email,
tape::{
MediaId,
drive::{

4
src/tape/pool_writer/mod.rs
View File

@ -13,16 +13,16 @@ use anyhow::{bail, Error};
use proxmox::tools::Uuid;
use pbs_datastore::task_log;
use pbs_tools::task_log;
use pbs_config::tape_encryption_keys::load_key_configs;
use pbs_tape::{
TapeWrite,
sg_tape::tape_alert_flags_critical,
};
use proxmox_rest_server::WorkerTask;
use crate::{
backup::{DataStore, SnapshotReader},
server::WorkerTask,
tape::{
TAPE_STATUS_DIR,
MAX_CHUNK_ARCHIVE_SIZE,