]> git.proxmox.com Git - proxmox-backup.git/blob - src/client/pull.rs
2555a14c93ebdbbec3854538678d29bfd867cfc2
[proxmox-backup.git] / src / client / pull.rs
1 //! Sync datastore from remote server
2
3 use anyhow::{bail, format_err, Error};
4 use serde_json::json;
5 use std::convert::TryFrom;
6 use std::sync::{Arc, Mutex};
7 use std::collections::{HashSet, HashMap};
8 use std::io::{Seek, SeekFrom};
9 use std::time::SystemTime;
10 use std::sync::atomic::{AtomicUsize, Ordering};
11
12 use proxmox::api::error::{StatusCode, HttpError};
13 use crate::{
14 tools::{ParallelHandler, compute_file_csum},
15 server::WorkerTask,
16 backup::*,
17 api2::types::*,
18 client::*,
19 };
20
21
22 // fixme: implement filters
23 // fixme: delete vanished groups
24 // Todo: correctly lock backup groups
25
26 async fn pull_index_chunks<I: IndexFile>(
27 worker: &WorkerTask,
28 chunk_reader: RemoteChunkReader,
29 target: Arc<DataStore>,
30 index: I,
31 downloaded_chunks: Arc<Mutex<HashSet<[u8;32]>>>,
32 ) -> Result<(), Error> {
33
34 use futures::stream::{self, StreamExt, TryStreamExt};
35
36 let start_time = SystemTime::now();
37
38 let stream = stream::iter(
39 (0..index.index_count())
40 .map(|pos| index.chunk_info(pos).unwrap())
41 .filter(|info| {
42 let mut guard = downloaded_chunks.lock().unwrap();
43 let done = guard.contains(&info.digest);
44 if !done {
45 // Note: We mark a chunk as downloaded before its actually downloaded
46 // to avoid duplicate downloads.
47 guard.insert(info.digest);
48 }
49 !done
50 })
51 );
52
53 let target2 = target.clone();
54 let verify_pool = ParallelHandler::new(
55 "sync chunk writer", 4,
56 move |(chunk, digest, size): (DataBlob, [u8;32], u64)| {
57 // println!("verify and write {}", proxmox::tools::digest_to_hex(&digest));
58 chunk.verify_unencrypted(size as usize, &digest)?;
59 target2.insert_chunk(&chunk, &digest)?;
60 Ok(())
61 }
62 );
63
64 let verify_and_write_channel = verify_pool.channel();
65
66 let bytes = Arc::new(AtomicUsize::new(0));
67
68 stream
69 .map(|info| {
70
71 let target = Arc::clone(&target);
72 let chunk_reader = chunk_reader.clone();
73 let bytes = Arc::clone(&bytes);
74 let verify_and_write_channel = verify_and_write_channel.clone();
75
76 Ok::<_, Error>(async move {
77 let chunk_exists = crate::tools::runtime::block_in_place(|| target.cond_touch_chunk(&info.digest, false))?;
78 if chunk_exists {
79 //worker.log(format!("chunk {} exists {}", pos, proxmox::tools::digest_to_hex(digest)));
80 return Ok::<_, Error>(());
81 }
82 //worker.log(format!("sync {} chunk {}", pos, proxmox::tools::digest_to_hex(digest)));
83 let chunk = chunk_reader.read_raw_chunk(&info.digest).await?;
84 let raw_size = chunk.raw_size() as usize;
85
86 // decode, verify and write in a separate threads to maximize throughput
87 crate::tools::runtime::block_in_place(|| verify_and_write_channel.send((chunk, info.digest, info.size())))?;
88
89 bytes.fetch_add(raw_size, Ordering::SeqCst);
90
91 Ok(())
92 })
93 })
94 .try_buffer_unordered(20)
95 .try_for_each(|_res| futures::future::ok(()))
96 .await?;
97
98 drop(verify_and_write_channel);
99
100 verify_pool.complete()?;
101
102 let elapsed = start_time.elapsed()?.as_secs_f64();
103
104 let bytes = bytes.load(Ordering::SeqCst);
105
106 worker.log(format!("downloaded {} bytes ({:.2} MiB/s)", bytes, (bytes as f64)/(1024.0*1024.0*elapsed)));
107
108 Ok(())
109 }
110
111 async fn download_manifest(
112 reader: &BackupReader,
113 filename: &std::path::Path,
114 ) -> Result<std::fs::File, Error> {
115
116 let mut tmp_manifest_file = std::fs::OpenOptions::new()
117 .write(true)
118 .create(true)
119 .truncate(true)
120 .read(true)
121 .open(&filename)?;
122
123 reader.download(MANIFEST_BLOB_NAME, &mut tmp_manifest_file).await?;
124
125 tmp_manifest_file.seek(SeekFrom::Start(0))?;
126
127 Ok(tmp_manifest_file)
128 }
129
130 fn verify_archive(
131 info: &FileInfo,
132 csum: &[u8; 32],
133 size: u64,
134 ) -> Result<(), Error> {
135 if size != info.size {
136 bail!("wrong size for file '{}' ({} != {})", info.filename, info.size, size);
137 }
138
139 if csum != &info.csum {
140 bail!("wrong checksum for file '{}'", info.filename);
141 }
142
143 Ok(())
144 }
145
146 async fn pull_single_archive(
147 worker: &WorkerTask,
148 reader: &BackupReader,
149 chunk_reader: &mut RemoteChunkReader,
150 tgt_store: Arc<DataStore>,
151 snapshot: &BackupDir,
152 archive_info: &FileInfo,
153 downloaded_chunks: Arc<Mutex<HashSet<[u8;32]>>>,
154 ) -> Result<(), Error> {
155
156 let archive_name = &archive_info.filename;
157 let mut path = tgt_store.base_path();
158 path.push(snapshot.relative_path());
159 path.push(archive_name);
160
161 let mut tmp_path = path.clone();
162 tmp_path.set_extension("tmp");
163
164 worker.log(format!("sync archive {}", archive_name));
165 let mut tmpfile = std::fs::OpenOptions::new()
166 .write(true)
167 .create(true)
168 .read(true)
169 .open(&tmp_path)?;
170
171 reader.download(archive_name, &mut tmpfile).await?;
172
173 match archive_type(archive_name)? {
174 ArchiveType::DynamicIndex => {
175 let index = DynamicIndexReader::new(tmpfile)
176 .map_err(|err| format_err!("unable to read dynamic index {:?} - {}", tmp_path, err))?;
177 let (csum, size) = index.compute_csum();
178 verify_archive(archive_info, &csum, size)?;
179
180 pull_index_chunks(worker, chunk_reader.clone(), tgt_store.clone(), index, downloaded_chunks).await?;
181 }
182 ArchiveType::FixedIndex => {
183 let index = FixedIndexReader::new(tmpfile)
184 .map_err(|err| format_err!("unable to read fixed index '{:?}' - {}", tmp_path, err))?;
185 let (csum, size) = index.compute_csum();
186 verify_archive(archive_info, &csum, size)?;
187
188 pull_index_chunks(worker, chunk_reader.clone(), tgt_store.clone(), index, downloaded_chunks).await?;
189 }
190 ArchiveType::Blob => {
191 let (csum, size) = compute_file_csum(&mut tmpfile)?;
192 verify_archive(archive_info, &csum, size)?;
193 }
194 }
195 if let Err(err) = std::fs::rename(&tmp_path, &path) {
196 bail!("Atomic rename file {:?} failed - {}", path, err);
197 }
198 Ok(())
199 }
200
201 // Note: The client.log.blob is uploaded after the backup, so it is
202 // not mentioned in the manifest.
203 async fn try_client_log_download(
204 worker: &WorkerTask,
205 reader: Arc<BackupReader>,
206 path: &std::path::Path,
207 ) -> Result<(), Error> {
208
209 let mut tmp_path = path.to_owned();
210 tmp_path.set_extension("tmp");
211
212 let tmpfile = std::fs::OpenOptions::new()
213 .write(true)
214 .create(true)
215 .read(true)
216 .open(&tmp_path)?;
217
218 // Note: be silent if there is no log - only log successful download
219 if let Ok(()) = reader.download(CLIENT_LOG_BLOB_NAME, tmpfile).await {
220 if let Err(err) = std::fs::rename(&tmp_path, &path) {
221 bail!("Atomic rename file {:?} failed - {}", path, err);
222 }
223 worker.log(format!("got backup log file {:?}", CLIENT_LOG_BLOB_NAME));
224 }
225
226 Ok(())
227 }
228
229 async fn pull_snapshot(
230 worker: &WorkerTask,
231 reader: Arc<BackupReader>,
232 tgt_store: Arc<DataStore>,
233 snapshot: &BackupDir,
234 downloaded_chunks: Arc<Mutex<HashSet<[u8;32]>>>,
235 ) -> Result<(), Error> {
236
237 let mut manifest_name = tgt_store.base_path();
238 manifest_name.push(snapshot.relative_path());
239 manifest_name.push(MANIFEST_BLOB_NAME);
240
241 let mut client_log_name = tgt_store.base_path();
242 client_log_name.push(snapshot.relative_path());
243 client_log_name.push(CLIENT_LOG_BLOB_NAME);
244
245 let mut tmp_manifest_name = manifest_name.clone();
246 tmp_manifest_name.set_extension("tmp");
247
248 let download_res = download_manifest(&reader, &tmp_manifest_name).await;
249 let mut tmp_manifest_file = match download_res {
250 Ok(manifest_file) => manifest_file,
251 Err(err) => {
252 match err.downcast_ref::<HttpError>() {
253 Some(HttpError { code, message }) => {
254 match *code {
255 StatusCode::NOT_FOUND => {
256 worker.log(format!("skipping snapshot {} - vanished since start of sync", snapshot));
257 return Ok(());
258 },
259 _ => {
260 bail!("HTTP error {} - {}", code, message);
261 },
262 }
263 },
264 None => {
265 return Err(err);
266 },
267 };
268 },
269 };
270 let tmp_manifest_blob = DataBlob::load_from_reader(&mut tmp_manifest_file)?;
271
272 if manifest_name.exists() {
273 let manifest_blob = proxmox::try_block!({
274 let mut manifest_file = std::fs::File::open(&manifest_name)
275 .map_err(|err| format_err!("unable to open local manifest {:?} - {}", manifest_name, err))?;
276
277 let manifest_blob = DataBlob::load_from_reader(&mut manifest_file)?;
278 Ok(manifest_blob)
279 }).map_err(|err: Error| {
280 format_err!("unable to read local manifest {:?} - {}", manifest_name, err)
281 })?;
282
283 if manifest_blob.raw_data() == tmp_manifest_blob.raw_data() {
284 if !client_log_name.exists() {
285 try_client_log_download(worker, reader, &client_log_name).await?;
286 }
287 worker.log("no data changes");
288 let _ = std::fs::remove_file(&tmp_manifest_name);
289 return Ok(()); // nothing changed
290 }
291 }
292
293 let manifest = BackupManifest::try_from(tmp_manifest_blob)?;
294
295 for item in manifest.files() {
296 let mut path = tgt_store.base_path();
297 path.push(snapshot.relative_path());
298 path.push(&item.filename);
299
300 if path.exists() {
301 match archive_type(&item.filename)? {
302 ArchiveType::DynamicIndex => {
303 let index = DynamicIndexReader::open(&path)?;
304 let (csum, size) = index.compute_csum();
305 match manifest.verify_file(&item.filename, &csum, size) {
306 Ok(_) => continue,
307 Err(err) => {
308 worker.log(format!("detected changed file {:?} - {}", path, err));
309 }
310 }
311 }
312 ArchiveType::FixedIndex => {
313 let index = FixedIndexReader::open(&path)?;
314 let (csum, size) = index.compute_csum();
315 match manifest.verify_file(&item.filename, &csum, size) {
316 Ok(_) => continue,
317 Err(err) => {
318 worker.log(format!("detected changed file {:?} - {}", path, err));
319 }
320 }
321 }
322 ArchiveType::Blob => {
323 let mut tmpfile = std::fs::File::open(&path)?;
324 let (csum, size) = compute_file_csum(&mut tmpfile)?;
325 match manifest.verify_file(&item.filename, &csum, size) {
326 Ok(_) => continue,
327 Err(err) => {
328 worker.log(format!("detected changed file {:?} - {}", path, err));
329 }
330 }
331 }
332 }
333 }
334
335 let mut chunk_reader = RemoteChunkReader::new(reader.clone(), None, item.chunk_crypt_mode(), HashMap::new());
336
337 pull_single_archive(
338 worker,
339 &reader,
340 &mut chunk_reader,
341 tgt_store.clone(),
342 snapshot,
343 &item,
344 downloaded_chunks.clone(),
345 ).await?;
346 }
347
348 if let Err(err) = std::fs::rename(&tmp_manifest_name, &manifest_name) {
349 bail!("Atomic rename file {:?} failed - {}", manifest_name, err);
350 }
351
352 if !client_log_name.exists() {
353 try_client_log_download(worker, reader, &client_log_name).await?;
354 }
355
356 // cleanup - remove stale files
357 tgt_store.cleanup_backup_dir(snapshot, &manifest)?;
358
359 Ok(())
360 }
361
362 pub async fn pull_snapshot_from(
363 worker: &WorkerTask,
364 reader: Arc<BackupReader>,
365 tgt_store: Arc<DataStore>,
366 snapshot: &BackupDir,
367 downloaded_chunks: Arc<Mutex<HashSet<[u8;32]>>>,
368 ) -> Result<(), Error> {
369
370 let (_path, is_new, _snap_lock) = tgt_store.create_locked_backup_dir(&snapshot)?;
371
372 if is_new {
373 worker.log(format!("sync snapshot {:?}", snapshot.relative_path()));
374
375 if let Err(err) = pull_snapshot(worker, reader, tgt_store.clone(), &snapshot, downloaded_chunks).await {
376 if let Err(cleanup_err) = tgt_store.remove_backup_dir(&snapshot, true) {
377 worker.log(format!("cleanup error - {}", cleanup_err));
378 }
379 return Err(err);
380 }
381 worker.log(format!("sync snapshot {:?} done", snapshot.relative_path()));
382 } else {
383 worker.log(format!("re-sync snapshot {:?}", snapshot.relative_path()));
384 pull_snapshot(worker, reader, tgt_store.clone(), &snapshot, downloaded_chunks).await?;
385 worker.log(format!("re-sync snapshot {:?} done", snapshot.relative_path()));
386 }
387
388 Ok(())
389 }
390
391 pub async fn pull_group(
392 worker: &WorkerTask,
393 client: &HttpClient,
394 src_repo: &BackupRepository,
395 tgt_store: Arc<DataStore>,
396 group: &BackupGroup,
397 delete: bool,
398 progress: &mut StoreProgress,
399 ) -> Result<(), Error> {
400
401 let path = format!("api2/json/admin/datastore/{}/snapshots", src_repo.store());
402
403 let args = json!({
404 "backup-type": group.backup_type(),
405 "backup-id": group.backup_id(),
406 });
407
408 let mut result = client.get(&path, Some(args)).await?;
409 let mut list: Vec<SnapshotListItem> = serde_json::from_value(result["data"].take())?;
410
411 list.sort_unstable_by(|a, b| a.backup_time.cmp(&b.backup_time));
412
413 client.login().await?; // make sure auth is complete
414
415 let fingerprint = client.fingerprint();
416
417 let last_sync = tgt_store.last_successful_backup(group)?;
418
419 let mut remote_snapshots = std::collections::HashSet::new();
420
421 // start with 16384 chunks (up to 65GB)
422 let downloaded_chunks = Arc::new(Mutex::new(HashSet::with_capacity(1024*64)));
423
424 progress.group_snapshots = list.len() as u64;
425
426 for (pos, item) in list.into_iter().enumerate() {
427 let snapshot = BackupDir::new(item.backup_type, item.backup_id, item.backup_time)?;
428
429 // in-progress backups can't be synced
430 if item.size.is_none() {
431 worker.log(format!("skipping snapshot {} - in-progress backup", snapshot));
432 continue;
433 }
434
435 let backup_time = snapshot.backup_time();
436
437 remote_snapshots.insert(backup_time);
438
439 if let Some(last_sync_time) = last_sync {
440 if last_sync_time > backup_time { continue; }
441 }
442
443 // get updated auth_info (new tickets)
444 let auth_info = client.login().await?;
445
446 let options = HttpClientOptions::new()
447 .password(Some(auth_info.ticket.clone()))
448 .fingerprint(fingerprint.clone());
449
450 let new_client = HttpClient::new(src_repo.host(), src_repo.port(), src_repo.auth_id(), options)?;
451
452 let reader = BackupReader::start(
453 new_client,
454 None,
455 src_repo.store(),
456 snapshot.group().backup_type(),
457 snapshot.group().backup_id(),
458 backup_time,
459 true,
460 ).await?;
461
462 let result = pull_snapshot_from(worker, reader, tgt_store.clone(), &snapshot, downloaded_chunks.clone()).await;
463
464 progress.done_snapshots = pos as u64 + 1;
465 worker.log(format!("percentage done: {}", progress.clone()));
466
467 result?; // stop on error
468 }
469
470 if delete {
471 let local_list = group.list_backups(&tgt_store.base_path())?;
472 for info in local_list {
473 let backup_time = info.backup_dir.backup_time();
474 if remote_snapshots.contains(&backup_time) { continue; }
475 worker.log(format!("delete vanished snapshot {:?}", info.backup_dir.relative_path()));
476 tgt_store.remove_backup_dir(&info.backup_dir, false)?;
477 }
478 }
479
480 Ok(())
481 }
482
483 pub async fn pull_store(
484 worker: &WorkerTask,
485 client: &HttpClient,
486 src_repo: &BackupRepository,
487 tgt_store: Arc<DataStore>,
488 delete: bool,
489 auth_id: Authid,
490 ) -> Result<(), Error> {
491
492 // explicit create shared lock to prevent GC on newly created chunks
493 let _shared_store_lock = tgt_store.try_shared_chunk_store_lock()?;
494
495 let path = format!("api2/json/admin/datastore/{}/groups", src_repo.store());
496
497 let mut result = client.get(&path, None).await?;
498
499 let mut list: Vec<GroupListItem> = serde_json::from_value(result["data"].take())?;
500
501 list.sort_unstable_by(|a, b| {
502 let type_order = a.backup_type.cmp(&b.backup_type);
503 if type_order == std::cmp::Ordering::Equal {
504 a.backup_id.cmp(&b.backup_id)
505 } else {
506 type_order
507 }
508 });
509
510 let mut errors = false;
511
512 let mut new_groups = std::collections::HashSet::new();
513 for item in list.iter() {
514 new_groups.insert(BackupGroup::new(&item.backup_type, &item.backup_id));
515 }
516
517 let mut progress = StoreProgress::new(list.len() as u64);
518
519 for (done, item) in list.into_iter().enumerate() {
520 progress.done_groups = done as u64;
521 progress.done_snapshots = 0;
522 progress.group_snapshots = 0;
523
524 let group = BackupGroup::new(&item.backup_type, &item.backup_id);
525
526 let (owner, _lock_guard) = match tgt_store.create_locked_backup_group(&group, &auth_id) {
527 Ok(result) => result,
528 Err(err) => {
529 worker.log(format!("sync group {}/{} failed - group lock failed: {}",
530 item.backup_type, item.backup_id, err));
531 errors = true; // do not stop here, instead continue
532 continue;
533 }
534 };
535
536 // permission check
537 if auth_id != owner { // only the owner is allowed to create additional snapshots
538 worker.log(format!("sync group {}/{} failed - owner check failed ({} != {})",
539 item.backup_type, item.backup_id, auth_id, owner));
540 errors = true; // do not stop here, instead continue
541
542 } else if let Err(err) = pull_group(
543 worker,
544 client,
545 src_repo,
546 tgt_store.clone(),
547 &group,
548 delete,
549 &mut progress,
550 ).await {
551 worker.log(format!(
552 "sync group {}/{} failed - {}",
553 item.backup_type,
554 item.backup_id,
555 err,
556 ));
557 errors = true; // do not stop here, instead continue
558 }
559 }
560
561 if delete {
562 let result: Result<(), Error> = proxmox::try_block!({
563 let local_groups = BackupInfo::list_backup_groups(&tgt_store.base_path())?;
564 for local_group in local_groups {
565 if new_groups.contains(&local_group) { continue; }
566 worker.log(format!("delete vanished group '{}/{}'", local_group.backup_type(), local_group.backup_id()));
567 if let Err(err) = tgt_store.remove_backup_group(&local_group) {
568 worker.log(err.to_string());
569 errors = true;
570 }
571 }
572 Ok(())
573 });
574 if let Err(err) = result {
575 worker.log(format!("error during cleanup: {}", err));
576 errors = true;
577 };
578 }
579
580 if errors {
581 bail!("sync failed with some errors.");
582 }
583
584 Ok(())
585 }