5 use crate::api2
::types
::*;
6 use crate::api_schema
::*;
7 use crate::api_schema
::router
::*;
8 //use crate::server::rest::*;
9 use serde_json
::{json, Value}
;
10 use std
::collections
::{HashSet, HashMap}
;
11 use chrono
::{DateTime, Datelike, TimeZone, Local}
;
12 use std
::path
::PathBuf
;
14 use proxmox
::tools
::{try_block, fs::file_get_contents, fs::file_set_contents}
;
16 use crate::config
::datastore
;
19 use crate::server
::WorkerTask
;
21 use hyper
::{header, Body, Response, StatusCode}
;
22 use hyper
::http
::request
::Parts
;
24 fn read_backup_index(store
: &DataStore
, backup_dir
: &BackupDir
) -> Result
<Value
, Error
> {
26 let mut path
= store
.base_path();
27 path
.push(backup_dir
.relative_path());
28 path
.push("index.json.blob");
30 let raw_data
= file_get_contents(&path
)?
;
31 let data
= DataBlob
::from_raw(raw_data
)?
.decode(None
)?
;
32 let mut result
: Value
= serde_json
::from_reader(&mut &data
[..])?
;
34 let result
= result
["files"].take();
36 if result
== Value
::Null
{
37 bail
!("missing 'files' property in backup index {:?}", path
);
43 fn group_backups(backup_list
: Vec
<BackupInfo
>) -> HashMap
<String
, Vec
<BackupInfo
>> {
45 let mut group_hash
= HashMap
::new();
47 for info
in backup_list
{
48 let group_id
= info
.backup_dir
.group().group_path().to_str().unwrap().to_owned();
49 let time_list
= group_hash
.entry(group_id
).or_insert(vec
![]);
56 fn mark_selections
<F
: Fn(DateTime
<Local
>, &BackupInfo
) -> String
> (
57 mark
: &mut HashSet
<PathBuf
>,
58 list
: &Vec
<BackupInfo
>,
62 let mut hash
= HashSet
::new();
64 let local_time
= info
.backup_dir
.backup_time().with_timezone(&Local
);
65 if hash
.len() >= keep
as usize { break; }
66 let backup_id
= info
.backup_dir
.relative_path();
67 let sel_id
: String
= select_id(local_time
, &info
);
68 if !hash
.contains(&sel_id
) {
70 //println!(" KEEP ID {} {}", backup_id, local_time.format("%c"));
71 mark
.insert(backup_id
);
79 _rpcenv
: &mut dyn RpcEnvironment
,
80 ) -> Result
<Value
, Error
> {
82 let store
= param
["store"].as_str().unwrap();
84 let datastore
= DataStore
::lookup_datastore(store
)?
;
86 let backup_list
= BackupInfo
::list_backups(&datastore
.base_path())?
;
88 let group_hash
= group_backups(backup_list
);
90 let mut groups
= vec
![];
92 for (_group_id
, mut list
) in group_hash
{
94 BackupInfo
::sort_list(&mut list
, false);
97 let group
= info
.backup_dir
.group();
100 "backup-type": group
.backup_type(),
101 "backup-id": group
.backup_id(),
102 "last-backup": info
.backup_dir
.backup_time().timestamp(),
103 "backup-count": list
.len() as u64,
111 fn list_snapshot_files (
114 _rpcenv
: &mut dyn RpcEnvironment
,
115 ) -> Result
<Value
, Error
> {
117 let store
= tools
::required_string_param(¶m
, "store")?
;
118 let backup_type
= tools
::required_string_param(¶m
, "backup-type")?
;
119 let backup_id
= tools
::required_string_param(¶m
, "backup-id")?
;
120 let backup_time
= tools
::required_integer_param(¶m
, "backup-time")?
;
122 let datastore
= DataStore
::lookup_datastore(store
)?
;
123 let snapshot
= BackupDir
::new(backup_type
, backup_id
, backup_time
);
125 let mut files
= read_backup_index(&datastore
, &snapshot
)?
;
127 let info
= BackupInfo
::new(&datastore
.base_path(), snapshot
)?
;
129 let file_set
= files
.as_array().unwrap().iter().fold(HashSet
::new(), |mut acc
, item
| {
130 acc
.insert(item
["filename"].as_str().unwrap().to_owned());
134 for file
in info
.files
{
135 if file_set
.contains(&file
) { continue; }
136 files
.as_array_mut().unwrap().push(json
!({ "filename": file }
));
142 fn delete_snapshots (
145 _rpcenv
: &mut dyn RpcEnvironment
,
146 ) -> Result
<Value
, Error
> {
148 let store
= tools
::required_string_param(¶m
, "store")?
;
149 let backup_type
= tools
::required_string_param(¶m
, "backup-type")?
;
150 let backup_id
= tools
::required_string_param(¶m
, "backup-id")?
;
151 let backup_time
= tools
::required_integer_param(¶m
, "backup-time")?
;
153 let snapshot
= BackupDir
::new(backup_type
, backup_id
, backup_time
);
155 let datastore
= DataStore
::lookup_datastore(store
)?
;
157 datastore
.remove_backup_dir(&snapshot
)?
;
165 _rpcenv
: &mut dyn RpcEnvironment
,
166 ) -> Result
<Value
, Error
> {
168 let store
= tools
::required_string_param(¶m
, "store")?
;
169 let backup_type
= param
["backup-type"].as_str();
170 let backup_id
= param
["backup-id"].as_str();
172 let datastore
= DataStore
::lookup_datastore(store
)?
;
174 let base_path
= datastore
.base_path();
176 let backup_list
= BackupInfo
::list_backups(&base_path
)?
;
178 let mut snapshots
= vec
![];
180 for info
in backup_list
{
181 let group
= info
.backup_dir
.group();
182 if let Some(backup_type
) = backup_type
{
183 if backup_type
!= group
.backup_type() { continue; }
185 if let Some(backup_id
) = backup_id
{
186 if backup_id
!= group
.backup_id() { continue; }
189 let mut result_item
= json
!({
190 "backup-type": group
.backup_type(),
191 "backup-id": group
.backup_id(),
192 "backup-time": info
.backup_dir
.backup_time().timestamp(),
196 if let Ok(index
) = read_backup_index(&datastore
, &info
.backup_dir
) {
197 let mut backup_size
= 0;
198 for item
in index
.as_array().unwrap().iter() {
199 if let Some(item_size
) = item
["size"].as_u64() {
200 backup_size
+= item_size
;
203 result_item
["size"] = backup_size
.into();
206 snapshots
.push(result_item
);
215 _rpcenv
: &mut dyn RpcEnvironment
,
216 ) -> Result
<Value
, Error
> {
218 let store
= param
["store"].as_str().unwrap();
220 let datastore
= DataStore
::lookup_datastore(store
)?
;
222 let base_path
= datastore
.base_path();
224 let mut stat
: libc
::statfs64
= unsafe { std::mem::zeroed() }
;
228 let res
= base_path
.with_nix_path(|cstr
| unsafe { libc::statfs64(cstr.as_ptr(), &mut stat) }
)?
;
229 nix
::errno
::Errno
::result(res
)?
;
231 let bsize
= stat
.f_bsize
as u64;
233 "total": stat
.f_blocks
*bsize
,
234 "used": (stat
.f_blocks
-stat
.f_bfree
)*bsize
,
235 "avail": stat
.f_bavail
*bsize
,
240 macro_rules
! add_common_prune_prameters
{
241 ($
( $list
:tt
)*) => {
246 &IntegerSchema
::new("Number of backups to keep.")
253 &IntegerSchema
::new("Number of daily backups to keep.")
260 &IntegerSchema
::new("Number of weekly backups to keep.")
267 &IntegerSchema
::new("Number of monthly backups to keep.")
274 &IntegerSchema
::new("Number of yearly backups to keep.")
283 const API_METHOD_STATUS
: ApiMethod
= ApiMethod
::new(
284 &ApiHandler
::Sync(&status
),
286 "Get datastore status.",
287 &add_common_prune_prameters
!(
288 ("store", false, &StringSchema
::new("Datastore name.").schema()),
297 _rpcenv
: &mut dyn RpcEnvironment
,
298 ) -> Result
<Value
, Error
> {
300 let store
= param
["store"].as_str().unwrap();
302 let backup_type
= tools
::required_string_param(¶m
, "backup-type")?
;
303 let backup_id
= tools
::required_string_param(¶m
, "backup-id")?
;
305 let group
= BackupGroup
::new(backup_type
, backup_id
);
307 let datastore
= DataStore
::lookup_datastore(store
)?
;
309 let mut keep_all
= true;
311 for opt
in &["keep-last", "keep-daily", "keep-weekly", "keep-weekly", "keep-yearly"] {
312 if !param
[opt
].is_null() {
318 let worker
= WorkerTask
::new("prune", Some(store
.to_owned()), "root@pam", true)?
;
319 let result
= try_block
! {
321 worker
.log("No prune selection - keeping all files.");
324 worker
.log(format
!("Starting prune on store {}", store
));
327 let mut list
= group
.list_backups(&datastore
.base_path())?
;
329 let mut mark
= HashSet
::new();
331 BackupInfo
::sort_list(&mut list
, false);
333 if let Some(keep_last
) = param
["keep-last"].as_u64() {
334 list
.iter().take(keep_last
as usize).for_each(|info
| {
335 mark
.insert(info
.backup_dir
.relative_path());
339 if let Some(keep_daily
) = param
["keep-daily"].as_u64() {
340 mark_selections(&mut mark
, &list
, keep_daily
as usize, |local_time
, _info
| {
341 format
!("{}/{}/{}", local_time
.year(), local_time
.month(), local_time
.day())
345 if let Some(keep_weekly
) = param
["keep-weekly"].as_u64() {
346 mark_selections(&mut mark
, &list
, keep_weekly
as usize, |local_time
, _info
| {
347 format
!("{}/{}", local_time
.year(), local_time
.iso_week().week())
351 if let Some(keep_monthly
) = param
["keep-monthly"].as_u64() {
352 mark_selections(&mut mark
, &list
, keep_monthly
as usize, |local_time
, _info
| {
353 format
!("{}/{}", local_time
.year(), local_time
.month())
357 if let Some(keep_yearly
) = param
["keep-yearly"].as_u64() {
358 mark_selections(&mut mark
, &list
, keep_yearly
as usize, |local_time
, _info
| {
359 format
!("{}/{}", local_time
.year(), local_time
.year())
363 let mut remove_list
: Vec
<BackupInfo
> = list
.into_iter()
364 .filter(|info
| !mark
.contains(&info
.backup_dir
.relative_path())).collect();
366 BackupInfo
::sort_list(&mut remove_list
, true);
368 for info
in remove_list
{
369 worker
.log(format
!("remove {:?}", info
.backup_dir
));
370 datastore
.remove_backup_dir(&info
.backup_dir
)?
;
376 worker
.log_result(&result
);
378 if let Err(err
) = result
{
379 bail
!("prune failed - {}", err
);
385 const API_METHOD_PRUNE
: ApiMethod
= ApiMethod
::new(
386 &ApiHandler
::Sync(&prune
),
388 "Prune the datastore.",
389 &add_common_prune_prameters
!(
390 ("store", false, &StringSchema
::new("Datastore name.").schema()),
391 ("backup-type", false, &BACKUP_TYPE_SCHEMA
),
392 ("backup-id", false, &BACKUP_ID_SCHEMA
),
397 fn start_garbage_collection(
400 rpcenv
: &mut dyn RpcEnvironment
,
401 ) -> Result
<Value
, Error
> {
403 let store
= param
["store"].as_str().unwrap().to_string();
405 let datastore
= DataStore
::lookup_datastore(&store
)?
;
407 println
!("Starting garbage collection on store {}", store
);
409 let to_stdout
= if rpcenv
.env_type() == RpcEnvironmentType
::CLI { true }
else { false }
;
411 let upid_str
= WorkerTask
::new_thread(
412 "garbage_collection", Some(store
.clone()), "root@pam", to_stdout
, move |worker
|
414 worker
.log(format
!("starting garbage collection on store {}", store
));
415 datastore
.garbage_collection(worker
)
421 pub const API_METHOD_START_GARBAGE_COLLECTION
: ApiMethod
= ApiMethod
::new(
422 &ApiHandler
::Sync(&start_garbage_collection
),
424 "Start garbage collection.",
425 &[ ("store", false, &StringSchema
::new("Datastore name.").schema()) ]
429 fn garbage_collection_status(
432 _rpcenv
: &mut dyn RpcEnvironment
,
433 ) -> Result
<Value
, Error
> {
435 let store
= param
["store"].as_str().unwrap();
437 let datastore
= DataStore
::lookup_datastore(&store
)?
;
439 println
!("Garbage collection status on store {}", store
);
441 let status
= datastore
.last_gc_status();
443 Ok(serde_json
::to_value(&status
)?
)
446 pub const API_METHOD_GARBAGE_COLLECTION_STATUS
: ApiMethod
= ApiMethod
::new(
447 &ApiHandler
::Sync(&garbage_collection_status
),
449 "Garbage collection status.",
450 &[ ("store", false, &StringSchema
::new("Datastore name.").schema()) ]
454 fn get_datastore_list(
457 _rpcenv
: &mut dyn RpcEnvironment
,
458 ) -> Result
<Value
, Error
> {
460 let config
= datastore
::config()?
;
462 Ok(config
.convert_to_array("store"))
471 _rpcenv
: Box
<dyn RpcEnvironment
>,
472 ) -> Result
<BoxFut
, Error
> {
474 let store
= tools
::required_string_param(¶m
, "store")?
;
476 let datastore
= DataStore
::lookup_datastore(store
)?
;
478 let file_name
= tools
::required_string_param(¶m
, "file-name")?
.to_owned();
480 let backup_type
= tools
::required_string_param(¶m
, "backup-type")?
;
481 let backup_id
= tools
::required_string_param(¶m
, "backup-id")?
;
482 let backup_time
= tools
::required_integer_param(¶m
, "backup-time")?
;
484 println
!("Download {} from {} ({}/{}/{}/{})", file_name
, store
,
485 backup_type
, backup_id
, Local
.timestamp(backup_time
, 0), file_name
);
487 let backup_dir
= BackupDir
::new(backup_type
, backup_id
, backup_time
);
489 let mut path
= datastore
.base_path();
490 path
.push(backup_dir
.relative_path());
491 path
.push(&file_name
);
493 let response_future
= tokio
::fs
::File
::open(path
)
494 .map_err(|err
| http_err
!(BAD_REQUEST
, format
!("File open failed: {}", err
)))
495 .and_then(move |file
| {
496 let payload
= tokio
::codec
::FramedRead
::new(file
, tokio
::codec
::BytesCodec
::new())
497 .map_ok(|bytes
| hyper
::Chunk
::from(bytes
.freeze()));
498 let body
= Body
::wrap_stream(payload
);
500 // fixme: set other headers ?
501 futures
::future
::ok(Response
::builder()
502 .status(StatusCode
::OK
)
503 .header(header
::CONTENT_TYPE
, "application/octet-stream")
508 Ok(Box
::new(response_future
))
511 pub const API_METHOD_DOWNLOAD_FILE
: ApiMethod
= ApiMethod
::new(
512 &ApiHandler
::Async(&download_file
),
514 "Download single raw file from backup snapshot.",
516 ("store", false, &StringSchema
::new("Datastore name.").schema()),
517 ("backup-type", false, &BACKUP_TYPE_SCHEMA
),
518 ("backup-id", false, &BACKUP_ID_SCHEMA
),
519 ("backup-time", false, &BACKUP_TIME_SCHEMA
),
520 ("file-name", false, &StringSchema
::new("Raw file name.")
521 .format(&FILENAME_FORMAT
)
528 fn upload_backup_log(
533 _rpcenv
: Box
<dyn RpcEnvironment
>,
534 ) -> Result
<BoxFut
, Error
> {
536 let store
= tools
::required_string_param(¶m
, "store")?
;
538 let datastore
= DataStore
::lookup_datastore(store
)?
;
540 let file_name
= "client.log.blob";
542 let backup_type
= tools
::required_string_param(¶m
, "backup-type")?
;
543 let backup_id
= tools
::required_string_param(¶m
, "backup-id")?
;
544 let backup_time
= tools
::required_integer_param(¶m
, "backup-time")?
;
546 let backup_dir
= BackupDir
::new(backup_type
, backup_id
, backup_time
);
548 let mut path
= datastore
.base_path();
549 path
.push(backup_dir
.relative_path());
550 path
.push(&file_name
);
553 bail
!("backup already contains a log.");
556 println
!("Upload backup log to {}/{}/{}/{}/{}", store
,
557 backup_type
, backup_id
, BackupDir
::backup_time_to_string(backup_dir
.backup_time()), file_name
);
560 .map_err(Error
::from
)
561 .try_fold(Vec
::new(), |mut acc
, chunk
| {
562 acc
.extend_from_slice(&*chunk
);
563 future
::ok
::<_
, Error
>(acc
)
565 .and_then(move |data
| async
move {
566 let blob
= DataBlob
::from_raw(data
)?
;
567 // always verify CRC at server side
569 let raw_data
= blob
.raw_data();
570 file_set_contents(&path
, raw_data
, None
)?
;
574 future
::ok(crate::server
::formatter
::json_response(Ok(Value
::Null
)))
581 pub const API_METHOD_UPLOAD_BACKUP_LOG
: ApiMethod
= ApiMethod
::new(
582 &ApiHandler
::Async(&upload_backup_log
),
584 "Download single raw file from backup snapshot.",
586 ("store", false, &StringSchema
::new("Datastore name.").schema()),
587 ("backup-type", false, &BACKUP_TYPE_SCHEMA
),
588 ("backup-id", false, &BACKUP_ID_SCHEMA
),
589 ("backup-time", false, &BACKUP_TIME_SCHEMA
),
594 const STORE_SCHEMA
: Schema
= StringSchema
::new("Datastore name.").schema();
596 const DATASTORE_INFO_SUBDIRS
: SubdirMap
= &[
600 .download(&API_METHOD_DOWNLOAD_FILE
)
607 &ApiHandler
::Sync(&list_snapshot_files
),
609 "List snapshot files.",
611 ("store", false, &STORE_SCHEMA
),
612 ("backup-type", false, &BACKUP_TYPE_SCHEMA
),
613 ("backup-id", false, &BACKUP_ID_SCHEMA
),
614 ("backup-time", false, &BACKUP_TIME_SCHEMA
),
623 .get(&API_METHOD_GARBAGE_COLLECTION_STATUS
)
624 .post(&API_METHOD_START_GARBAGE_COLLECTION
)
631 &ApiHandler
::Sync(&list_groups
),
633 "List backup groups.",
634 &[ ("store", false, &STORE_SCHEMA
) ],
642 .post(&API_METHOD_PRUNE
)
649 &ApiHandler
::Sync(&list_snapshots
),
651 "List backup groups.",
653 ("store", false, &STORE_SCHEMA
),
654 ("backup-type", true, &BACKUP_TYPE_SCHEMA
),
655 ("backup-id", true, &BACKUP_ID_SCHEMA
),
662 &ApiHandler
::Sync(&delete_snapshots
),
664 "Delete backup snapshot.",
666 ("store", false, &STORE_SCHEMA
),
667 ("backup-type", false, &BACKUP_TYPE_SCHEMA
),
668 ("backup-id", false, &BACKUP_ID_SCHEMA
),
669 ("backup-time", false, &BACKUP_TIME_SCHEMA
),
678 .get(&API_METHOD_STATUS
)
683 .upload(&API_METHOD_UPLOAD_BACKUP_LOG
)
687 const DATASTORE_INFO_ROUTER
: Router
= Router
::new()
688 .get(&list_subdirs_api_method
!(DATASTORE_INFO_SUBDIRS
))
689 .subdirs(DATASTORE_INFO_SUBDIRS
);
692 pub const ROUTER
: Router
= Router
::new()
695 &ApiHandler
::Sync(&get_datastore_list
),
696 &ObjectSchema
::new("Directory index.", &[])
699 .match_all("store", &DATASTORE_INFO_ROUTER
);