src/api2/pull.rs

   1 //! Sync datastore from remote server
   2 use std::sync::{Arc};
   3
   4 use anyhow::{format_err, Error};
   5 use futures::{select, future::FutureExt};
   6
   7 use proxmox::api::api;
   8 use proxmox::api::{ApiMethod, Router, RpcEnvironment, Permission};
   9
  10 use pbs_client::{HttpClient, BackupRepository};
  11 use pbs_api_types::{
  12     Remote, Authid, SyncJobConfig,
  13     DATASTORE_SCHEMA, REMOTE_ID_SCHEMA, REMOVE_VANISHED_BACKUPS_SCHEMA,
  14     PRIV_DATASTORE_BACKUP, PRIV_DATASTORE_PRUNE, PRIV_REMOTE_READ,
  15 };
  16
  17 use crate::server::{WorkerTask, jobstate::Job, pull::pull_store};
  18 use crate::backup::DataStore;
  19 use crate::config::cached_user_info::CachedUserInfo;
  20
  21 pub fn check_pull_privs(
  22     auth_id: &Authid,
  23     store: &str,
  24     remote: &str,
  25     remote_store: &str,
  26     delete: bool,
  27 ) -> Result<(), Error> {
  28
  29     let user_info = CachedUserInfo::new()?;
  30
  31     user_info.check_privs(auth_id, &["datastore", store], PRIV_DATASTORE_BACKUP, false)?;
  32     user_info.check_privs(auth_id, &["remote", remote, remote_store], PRIV_REMOTE_READ, false)?;
  33
  34     if delete {
  35         user_info.check_privs(auth_id, &["datastore", store], PRIV_DATASTORE_PRUNE, false)?;
  36     }
  37
  38     Ok(())
  39 }
  40
  41 pub async fn get_pull_parameters(
  42     store: &str,
  43     remote: &str,
  44     remote_store: &str,
  45 ) -> Result<(HttpClient, BackupRepository, Arc<DataStore>), Error> {
  46
  47     let tgt_store = DataStore::lookup_datastore(store)?;
  48
  49     let (remote_config, _digest) = pbs_config::remote::config()?;
  50     let remote: Remote = remote_config.lookup("remote", remote)?;
  51
  52     let src_repo = BackupRepository::new(
  53         Some(remote.config.auth_id.clone()),
  54         Some(remote.config.host.clone()),
  55         remote.config.port,
  56         remote_store.to_string(),
  57     );
  58
  59     let client = crate::api2::config::remote::remote_client(remote).await?;
  60
  61     Ok((client, src_repo, tgt_store))
  62 }
  63
  64 pub fn do_sync_job(
  65     mut job: Job,
  66     sync_job: SyncJobConfig,
  67     auth_id: &Authid,
  68     schedule: Option<String>,
  69 ) -> Result<String, Error> {
  70
  71     let job_id = format!("{}:{}:{}:{}",
  72                          sync_job.remote,
  73                          sync_job.remote_store,
  74                          sync_job.store,
  75                          job.jobname());
  76     let worker_type = job.jobtype().to_string();
  77
  78     let (email, notify) = crate::server::lookup_datastore_notify_settings(&sync_job.store);
  79
  80     let upid_str = WorkerTask::spawn(
  81         &worker_type,
  82         Some(job_id.clone()),
  83         auth_id.clone(),
  84         false,
  85         move |worker| async move {
  86
  87             job.start(&worker.upid().to_string())?;
  88
  89             let worker2 = worker.clone();
  90             let sync_job2 = sync_job.clone();
  91
  92             let worker_future = async move {
  93
  94                 let delete = sync_job.remove_vanished.unwrap_or(true);
  95                 let sync_owner = sync_job.owner.unwrap_or_else(|| Authid::root_auth_id().clone());
  96                 let (client, src_repo, tgt_store) = get_pull_parameters(&sync_job.store, &sync_job.remote, &sync_job.remote_store).await?;
  97
  98                 worker.log(format!("Starting datastore sync job '{}'", job_id));
  99                 if let Some(event_str) = schedule {
 100                     worker.log(format!("task triggered by schedule '{}'", event_str));
 101                 }
 102                 worker.log(format!("Sync datastore '{}' from '{}/{}'",
 103                         sync_job.store, sync_job.remote, sync_job.remote_store));
 104
 105                 pull_store(&worker, &client, &src_repo, tgt_store.clone(), delete, sync_owner).await?;
 106
 107                 worker.log(format!("sync job '{}' end", &job_id));
 108
 109                 Ok(())
 110             };
 111
 112             let mut abort_future = worker2.abort_future().map(|_| Err(format_err!("sync aborted")));
 113
 114             let result = select!{
 115                 worker = worker_future.fuse() => worker,
 116                 abort = abort_future => abort,
 117             };
 118
 119             let status = worker2.create_state(&result);
 120
 121             match job.finish(status) {
 122                 Ok(_) => {},
 123                 Err(err) => {
 124                     eprintln!("could not finish job state: {}", err);
 125                 }
 126             }
 127
 128             if let Some(email) = email {
 129                 if let Err(err) = crate::server::send_sync_status(&email, notify, &sync_job2, &result) {
 130                     eprintln!("send sync notification failed: {}", err);
 131                 }
 132             }
 133
 134             result
 135         })?;
 136
 137     Ok(upid_str)
 138 }
 139
 140 #[api(
 141     input: {
 142         properties: {
 143             store: {
 144                 schema: DATASTORE_SCHEMA,
 145             },
 146             remote: {
 147                 schema: REMOTE_ID_SCHEMA,
 148             },
 149             "remote-store": {
 150                 schema: DATASTORE_SCHEMA,
 151             },
 152             "remove-vanished": {
 153                 schema: REMOVE_VANISHED_BACKUPS_SCHEMA,
 154                 optional: true,
 155             },
 156         },
 157     },
 158     access: {
 159         // Note: used parameters are no uri parameters, so we need to test inside function body
 160         description: r###"The user needs Datastore.Backup privilege on '/datastore/{store}',
 161 and needs to own the backup group. Remote.Read is required on '/remote/{remote}/{remote-store}'.
 162 The delete flag additionally requires the Datastore.Prune privilege on '/datastore/{store}'.
 163 "###,
 164         permission: &Permission::Anybody,
 165     },
 166 )]
 167 /// Sync store from other repository
 168 async fn pull (
 169     store: String,
 170     remote: String,
 171     remote_store: String,
 172     remove_vanished: Option<bool>,
 173     _info: &ApiMethod,
 174     rpcenv: &mut dyn RpcEnvironment,
 175 ) -> Result<String, Error> {
 176
 177     let auth_id: Authid = rpcenv.get_auth_id().unwrap().parse()?;
 178     let delete = remove_vanished.unwrap_or(true);
 179
 180     check_pull_privs(&auth_id, &store, &remote, &remote_store, delete)?;
 181
 182     let (client, src_repo, tgt_store) = get_pull_parameters(&store, &remote, &remote_store).await?;
 183
 184     // fixme: set to_stdout to false?
 185     let upid_str = WorkerTask::spawn("sync", Some(store.clone()), auth_id.clone(), true, move |worker| async move {
 186
 187         worker.log(format!("sync datastore '{}' start", store));
 188
 189         let pull_future = pull_store(&worker, &client, &src_repo, tgt_store.clone(), delete, auth_id);
 190         let future = select!{
 191             success = pull_future.fuse() => success,
 192             abort = worker.abort_future().map(|_| Err(format_err!("pull aborted"))) => abort,
 193         };
 194
 195         let _ = future?;
 196
 197         worker.log(format!("sync datastore '{}' end", store));
 198
 199         Ok(())
 200     })?;
 201
 202     Ok(upid_str)
 203 }
 204
 205 pub const ROUTER: Router = Router::new()
 206     .post(&API_METHOD_PULL);