From 2217b00a2caf5982e09f1d3db7e3e21347d88bae Mon Sep 17 00:00:00 2001 From: Tim Bruijnzeels Date: Tue, 29 Sep 2020 09:05:46 +0200 Subject: [PATCH] Optionally archive old events (#307) --- Changelog.md | 32 +++++++ defaults/krill.conf | 28 ++++++ src/commons/api/history.rs | 18 +++- src/commons/eventsourcing/agg_store.rs | 59 +++++++++++- src/commons/eventsourcing/mod.rs | 2 +- src/commons/eventsourcing/store.rs | 6 ++ src/daemon/ca/server.rs | 12 ++- src/daemon/config.rs | 12 +++ src/daemon/http/server.rs | 20 ++-- src/daemon/krillserver.rs | 73 +++++++-------- src/daemon/scheduler.rs | 32 ++++++- src/pubd/pubserver.rs | 124 +++++++++++++------------ src/publish/mod.rs | 8 +- test-resources/krill-init.conf | 28 ++++++ 14 files changed, 333 insertions(+), 121 deletions(-) diff --git a/Changelog.md b/Changelog.md index 896e317d..5c011252 100644 --- a/Changelog.md +++ b/Changelog.md @@ -5,6 +5,38 @@ for planned releases. ## 0.8.0 RC + +### Archiving old events + +You can now choose to archive old publication events which will allow you to +delete or move them in order to save space on your system. To use this feature, +which is disabled by default, you need to set the following directive in your +config file: + +``` + archive_threshold_days = +``` + +If enabled this option will make sure that the republish events, where your CA +simply generates a new Manifest and CRL are archived after the given number of +days. + +If you run Krill as a Publication Server then this option will enable the archiving +of publication deltas received by your server from CAs. + +Archived commands (containing e.g. details of when the change happened), and following +events will be moved to an "archived" subdirectory under your data directory as follows: +``` + $data_dir/pubd/0/archived <-- if you run Krill as a Publication Server + $data_dir/cas/ca/archived <-- for a CA named 'ca' +``` + +If you want to save space you can delete the files from these archived directories, +e.g. from cron. However, you could also archive them in a different way: e.g. compress +and move to long term storage. Krill will no longer need this data, but if you ever wanted +to see these details in history you would need to move them back into the parent directory +of the 'archived' directory. + Features: * Will check all state on startup and try to recover in case of issues (#314) diff --git a/defaults/krill.conf b/defaults/krill.conf index 0916895f..8113f4eb 100644 --- a/defaults/krill.conf +++ b/defaults/krill.conf @@ -22,6 +22,34 @@ # ### data_dir = "./data" +# Archive publication events after X days. If you do NOT set this +# value then Krill will not do any archiving. +# +# If enabled this option will make sure that the republish events, +# where your CA simply generates a new Manifest and CRL are archived +# after the given number of days. +# +# If you run Krill as a Publication Server then this option will +# enable the archiving of publication deltas received by your server +# from CAs. +# +# Archived commands (containing e.g. details of when the change happened), +# and following events will be moved to an "archived" subdirectory under +# your data directory as follows: +# $data_dir/pubd/0/archived <-- for Publication Server +# $data_dir/cas/ca/archived <-- for a CA named 'ca' +# +# If you want to save space you can delete the files from these archived +# directories, e.g. from cron. However, you could also archive them in +# a different way: e.g. compress and move to long term storage. Krill will +# no longer need this data, but if you ever wanted to see these details in +# history you would need to move them back into the parent directory of +# the 'archived' directory. +# +# To enable this set the following key value pair. +# +### archive_threshold_days = 7 + # Specify the path to the PID file for Krill. # # Defaults to "krill.pid" under the 'data_dir' specified above. diff --git a/src/commons/api/history.rs b/src/commons/api/history.rs index 997bfa7a..b9243377 100644 --- a/src/commons/api/history.rs +++ b/src/commons/api/history.rs @@ -155,6 +155,18 @@ impl CommandHistoryRecord { Time::from(DateTime::from_utc(time, Utc)) } + pub fn resulting_version(&self) -> u64 { + if let Some(versions) = self.effect.events() { + if let Some(last) = versions.last() { + *last + } else { + self.version + } + } else { + self.version + } + } + pub fn command_key(&self) -> Result { CommandKey::from_str(&self.key) } @@ -290,10 +302,14 @@ pub struct CommandHistoryCriteria { } impl CommandHistoryCriteria { - pub fn set_exclude(&mut self, labels: &[&str]) { + pub fn set_excludes(&mut self, labels: &[&str]) { self.label_excludes = Some(labels.iter().map(|s| (*s).to_string()).collect()); } + pub fn set_includes(&mut self, labels: &[&str]) { + self.label_includes = Some(labels.iter().map(|s| (*s).to_string()).collect()); + } + pub fn set_after(&mut self, timestamp: i64) { self.after = Some(timestamp); } diff --git a/src/commons/eventsourcing/agg_store.rs b/src/commons/eventsourcing/agg_store.rs index 27ad3c85..976a2838 100644 --- a/src/commons/eventsourcing/agg_store.rs +++ b/src/commons/eventsourcing/agg_store.rs @@ -3,6 +3,7 @@ use std::io; use std::path::PathBuf; use std::sync::{Arc, RwLock}; +use chrono::Duration; use rpki::x509::Time; use crate::commons::api::{CommandHistory, CommandHistoryCriteria, Handle}; @@ -90,6 +91,9 @@ pub enum AggregateStoreError { #[display(fmt = "Could not recover state for '{}', aborting recover. Use backup!!", _0)] CouldNotRecover(Handle), + + #[display(fmt = "Could not archive commands and events for '{}'. Error: {}", _0, _1)] + CouldNotArchive(Handle, String), } impl From for AggregateStoreError { @@ -136,8 +140,15 @@ where /// In that case the user may want to use the recover option to see what can be salvaged. pub fn warm(&self) -> StoreResult<()> { for handle in self.list() { - self.get_latest(&handle) - .map_err(|e| AggregateStoreError::WarmupFailed(handle, e.to_string()))?; + if !self.store.has_key(&handle, &DiskKeyStore::key_for_snapshot()) { + return Err(AggregateStoreError::WarmupFailed( + handle, + "No snapshot found.".to_string(), + )); + } else { + self.get_latest(&handle) + .map_err(|e| AggregateStoreError::WarmupFailed(handle, e.to_string()))?; + } } Ok(()) } @@ -233,6 +244,50 @@ where Ok(()) } + + /// Archive old commands if they are: + /// - older than the backup snapshot + /// - AND older then the threshold days + /// - AND they are eligible for archiving + pub fn archive_old_commands(&self, handle: &Handle, days: i64) -> StoreResult<()> { + let mut crit = CommandHistoryCriteria::default(); + let before = (Time::now() - Duration::days(days)).timestamp(); + crit.set_before(before); + crit.set_includes(&["cmd-ca-publish", "pubd-publish"]); + + let info = self + .store + .get_info(handle) + .map_err(|e| AggregateStoreError::CouldNotArchive(handle.clone(), e.to_string()))?; + + let archivable = self.command_history(handle, crit)?; + + let commands = archivable.commands(); + + for command in commands { + let key = command + .command_key() + .map_err(|e| AggregateStoreError::CouldNotArchive(handle.clone(), e.to_string()))?; + + if command.resulting_version() < info.snapshot_version { + info!("Archiving command {} for {}", command.key, handle); + + self.store + .archive(handle, &key.into()) + .map_err(|e| AggregateStoreError::CouldNotArchive(handle.clone(), e.to_string()))?; + + if let Some(evt_versions) = command.effect.events() { + for version in evt_versions { + info!("Archiving event {} for {}", version, handle); + self.store + .archive_event(handle, *version) + .map_err(|e| AggregateStoreError::CouldNotArchive(handle.clone(), e.to_string()))?; + } + } + } + } + Ok(()) + } } impl DiskAggregateStore diff --git a/src/commons/eventsourcing/mod.rs b/src/commons/eventsourcing/mod.rs index 35befff1..a95af955 100644 --- a/src/commons/eventsourcing/mod.rs +++ b/src/commons/eventsourcing/mod.rs @@ -329,7 +329,7 @@ mod tests { // Get history excluding 'around the sun' commands let mut crit = CommandHistoryCriteria::default(); - crit.set_exclude(&["person-around-sun"]); + crit.set_excludes(&["person-around-sun"]); let history = manager.command_history(&id_alice, crit).unwrap(); assert_eq!(history.total(), 1); diff --git a/src/commons/eventsourcing/store.rs b/src/commons/eventsourcing/store.rs index df8cde04..55c33584 100644 --- a/src/commons/eventsourcing/store.rs +++ b/src/commons/eventsourcing/store.rs @@ -123,6 +123,12 @@ impl TryFrom for CommandKey { #[derive(Clone, Debug, Eq, PartialEq)] pub struct CommandKeyError; +impl fmt::Display for CommandKeyError { + fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { + write!(f, "invalid command key") + } +} + //------------ KeyStore ------------------------------------------------------ /// Generic KeyStore for AggregateManager diff --git a/src/daemon/ca/server.rs b/src/daemon/ca/server.rs index e24d9a38..5ae1036c 100644 --- a/src/daemon/ca/server.rs +++ b/src/daemon/ca/server.rs @@ -24,7 +24,7 @@ use crate::commons::eventsourcing::{Aggregate, AggregateStore, Command, CommandK use crate::commons::remote::cmslogger::CmsLogger; use crate::commons::remote::{rfc6492, rfc8181, rfc8183}; use crate::commons::util::httpclient; -use crate::commons::KrillResult; +use crate::commons::{KrillEmptyResult, KrillResult}; use crate::constants::{CASERVER_DIR, STATUS_DIR}; use crate::daemon::ca::{ self, ta_handle, CertAuth, Cmd, CmdDet, IniDet, ResourceTaggedAttestation, RouteAuthorizationUpdates, @@ -376,6 +376,16 @@ impl CaServer { .map_err(|_| Error::CaUnknown(handle.clone())) } + /// Archive old (eligible) commands for CA + pub async fn archive_ca_commands(&self, days: i64) -> KrillEmptyResult { + for ca in self.ca_list().await.cas() { + let lock = self.locks.ca(ca.handle()).await; + let _ = lock.write().await; + self.ca_store.archive_old_commands(ca.handle(), days)?; + } + Ok(()) + } + /// Shows the details for a CA command pub fn get_ca_command_details( &self, diff --git a/src/daemon/config.rs b/src/daemon/config.rs index 521caf98..664bd456 100644 --- a/src/daemon/config.rs +++ b/src/daemon/config.rs @@ -61,6 +61,13 @@ impl ConfigDefaults { fn data_dir() -> PathBuf { PathBuf::from("./data") } + fn archive_threshold_days() -> Option { + if Self::test_mode() { + Some(0) + } else { + None + } + } fn rsync_base() -> uri::Rsync { uri::Rsync::from_str("rsync://localhost/repo/").unwrap() } @@ -206,6 +213,9 @@ pub struct Config { #[serde(default = "ConfigDefaults::data_dir")] pub data_dir: PathBuf, + #[serde(default = "ConfigDefaults::archive_threshold_days")] + pub archive_threshold_days: Option, + pub pid_file: Option, #[serde(default = "ConfigDefaults::rsync_base")] @@ -359,6 +369,7 @@ impl Config { let testbed_enabled = true; let https_mode = HttpsMode::Generate; let data_dir = data_dir.clone(); + let archive_threshold_days = Some(0); let rsync_base = ConfigDefaults::rsync_base(); let service_uri = ConfigDefaults::service_uri(); let rrdp_service_uri = Some("https://localhost:3000/test-rrdp/".to_string()); @@ -409,6 +420,7 @@ impl Config { testbed_enabled, https_mode, data_dir, + archive_threshold_days, rsync_base, service_uri, rrdp_service_uri, diff --git a/src/daemon/http/server.rs b/src/daemon/http/server.rs index aa850e42..6033a06c 100644 --- a/src/daemon/http/server.rs +++ b/src/daemon/http/server.rs @@ -237,7 +237,7 @@ pub async fn metrics(req: Request) -> RoutingResult { res.push_str(&format!("krill_version_patch {}\n", KRILL_VERSION_PATCH)); res.push_str("\n"); - if let Ok(stats) = server.repo_stats().await { + if let Ok(stats) = server.repo_stats() { let publishers = stats.get_publishers(); res.push_str("# HELP krill_repo_publisher number of publishers in repository\n"); @@ -435,7 +435,7 @@ pub async fn rfc8181(req: Request) -> RoutingResult { }; let read = state.read().await; - match read.rfc8181(publisher, bytes).await { + match read.rfc8181(publisher, bytes) { Ok(bytes) => Ok(HttpResponse::rfc8181(bytes.to_vec())), Err(e) => render_error(e), } @@ -504,7 +504,7 @@ async fn stats(req: Request) -> RoutingResult { match *req.method() { Method::GET => match req.path().full() { "/stats/info" => render_json(req.state().read().await.server_info()), - "/stats/repo" => render_json_res(req.state().read().await.repo_stats().await), + "/stats/repo" => render_json_res(req.state().read().await.repo_stats()), "/stats/cas" => render_json(req.state().read().await.cas_stats().await), _ => Err(req), }, @@ -674,7 +674,6 @@ pub async fn stale_publishers(req: Request, seconds: Option<&str>) -> RoutingRes .read() .await .repo_stats() - .await .map(|stats| PublisherList::build(&stats.stale_publishers(seconds), "/api/v1/publishers")), ), Err(_) => render_error(Error::ApiInvalidSeconds), @@ -688,7 +687,6 @@ pub async fn list_pbl(req: Request) -> RoutingResult { .read() .await .publishers() - .await .map(|publishers| PublisherList::build(&publishers, "/api/v1/publishers")), ) } @@ -697,22 +695,20 @@ pub async fn list_pbl(req: Request) -> RoutingResult { async fn add_pbl(req: Request) -> RoutingResult { let server = req.state().clone(); match req.json().await { - Ok(pbl) => render_json_res(server.write().await.add_publisher(pbl).await), + Ok(pbl) => render_json_res(server.write().await.add_publisher(pbl)), Err(e) => render_error(e), } } /// Removes a publisher. Should be idempotent! If if did not exist then /// that's just fine. -#[allow(clippy::needless_pass_by_value)] pub async fn remove_pbl(req: Request, publisher: Handle) -> RoutingResult { - render_empty_res(req.state().write().await.remove_publisher(publisher).await) + render_empty_res(req.state().write().await.remove_publisher(publisher)) } /// Returns a json structure with publisher details -#[allow(clippy::needless_pass_by_value)] pub async fn show_pbl(req: Request, publisher: Handle) -> RoutingResult { - render_json_res(req.state().read().await.get_publisher(&publisher).await) + render_json_res(req.state().read().await.get_publisher(&publisher)) } //------------ repository_response --------------------------------------------- @@ -732,7 +728,7 @@ pub async fn repository_response_json(req: Request, publisher: Handle) -> Routin } async fn repository_response(req: &Request, publisher: &Handle) -> Result { - req.state().read().await.repository_response(publisher).await + req.state().read().await.repository_response(publisher) } async fn ca_add_child(req: Request, parent: ParentHandle) -> RoutingResult { @@ -867,7 +863,7 @@ fn parse_history_path(path: &mut RequestPath) -> Option let mut crit = CommandHistoryCriteria::default(); match path.next() { - Some("short") => crit.set_exclude(&["cmd-ca-publish"]), + Some("short") => crit.set_excludes(&["cmd-ca-publish"]), Some("full") => {} _ => return None, }; diff --git a/src/daemon/krillserver.rs b/src/daemon/krillserver.rs index ae7f1f4b..a5fbdf5d 100644 --- a/src/daemon/krillserver.rs +++ b/src/daemon/krillserver.rs @@ -114,16 +114,13 @@ impl KrillServer { let pubserver = { if CONFIG.repo_enabled { - Some( - PubServer::build( - &base_uri, - rrdp_base_uri.clone(), - work_dir, - CONFIG.rfc8181_log_dir.as_ref(), - Arc::new(signer.clone()), - ) - .await?, - ) + Some(PubServer::build( + &base_uri, + rrdp_base_uri.clone(), + work_dir, + CONFIG.rfc8181_log_dir.as_ref(), + Arc::new(signer.clone()), + )?) } else { PubServer::remove_if_empty( &base_uri, @@ -131,8 +128,7 @@ impl KrillServer { work_dir, CONFIG.rfc8181_log_dir.as_ref(), Arc::new(signer.clone()), - ) - .await? + )? } }; let pubserver: Option> = pubserver.map(Arc::new); @@ -155,7 +151,7 @@ impl KrillServer { info!("Creating embedded Trust Anchor"); let pubserver = pubserver.as_ref().ok_or_else(|| Error::PublisherNoEmbeddedRepo)?; - let repo_info: RepoInfo = pubserver.repo_info_for(&ta_handle).await?; + let repo_info: RepoInfo = pubserver.repo_info_for(&ta_handle)?; let ta_uri = CONFIG.ta_cert_uri(); @@ -170,7 +166,7 @@ impl KrillServer { // Add publisher let req = rfc8183::PublisherRequest::new(None, ta_handle.clone(), ta.id_cert().clone()); - pubserver.create_publisher(req).await?; + pubserver.create_publisher(req)?; // Force initial publication caserver.republish(&ta_handle).await?; @@ -192,10 +188,10 @@ impl KrillServer { let pubserver = pubserver.as_ref().ok_or_else(|| Error::PublisherNoEmbeddedRepo)?; let pub_req = rfc8183::PublisherRequest::new(None, testbed_ca_handle.clone(), testbed_ca.id_cert().clone()); - pubserver.create_publisher(pub_req).await?; + pubserver.create_publisher(pub_req)?; let rfc8181_uri = uri::Https::from_string(format!("{}rfc8181/{}", service_uri, testbed_ca_handle)).unwrap(); - let repo_response = pubserver.repository_response(rfc8181_uri, &testbed_ca_handle).await?; + let repo_response = pubserver.repository_response(rfc8181_uri, &testbed_ca_handle)?; let repo_contact = RepositoryContact::Rfc8181(repo_response); caserver.update_repo(testbed_ca_handle.clone(), repo_contact).await?; caserver.republish(&testbed_ca_handle).await?; @@ -282,32 +278,32 @@ impl KrillServer { } /// Returns the repository server stats - pub async fn repo_stats(&self) -> KrillResult { - self.get_embedded()?.repo_stats().await + pub fn repo_stats(&self) -> KrillResult { + self.get_embedded()?.repo_stats() } /// Returns all currently CONFIGured publishers. (excludes deactivated) - pub async fn publishers(&self) -> KrillResult> { - self.get_embedded()?.publishers().await + pub fn publishers(&self) -> KrillResult> { + self.get_embedded()?.publishers() } /// Adds the publishers, blows up if it already existed. - pub async fn add_publisher(&self, req: rfc8183::PublisherRequest) -> KrillResult { + pub fn add_publisher(&self, req: rfc8183::PublisherRequest) -> KrillResult { let publisher_handle = req.publisher_handle().clone(); - self.get_embedded()?.create_publisher(req).await?; + self.get_embedded()?.create_publisher(req)?; - self.repository_response(&publisher_handle).await + self.repository_response(&publisher_handle) } /// Removes a publisher, blows up if it didn't exist. - pub async fn remove_publisher(&mut self, publisher: PublisherHandle) -> KrillEmptyResult { - self.get_embedded()?.remove_publisher(publisher).await + pub fn remove_publisher(&mut self, publisher: PublisherHandle) -> KrillEmptyResult { + self.get_embedded()?.remove_publisher(publisher) } /// Returns a publisher. - pub async fn get_publisher(&self, publisher: &PublisherHandle) -> KrillResult { - self.get_embedded()?.get_publisher_details(publisher).await + pub fn get_publisher(&self, publisher: &PublisherHandle) -> KrillResult { + self.get_embedded()?.get_publisher_details(publisher) } pub fn rrdp_base_path(&self) -> PathBuf { @@ -320,14 +316,13 @@ impl KrillServer { /// # Manage RFC8181 clients /// impl KrillServer { - pub async fn repository_response(&self, publisher: &PublisherHandle) -> KrillResult { + pub fn repository_response(&self, publisher: &PublisherHandle) -> KrillResult { let rfc8181_uri = uri::Https::from_string(format!("{}rfc8181/{}", self.service_uri, publisher)).unwrap(); - - self.get_embedded()?.repository_response(rfc8181_uri, publisher).await + self.get_embedded()?.repository_response(rfc8181_uri, publisher) } - pub async fn rfc8181(&self, publisher: PublisherHandle, msg_bytes: Bytes) -> KrillResult { - self.get_embedded()?.rfc8181(publisher, msg_bytes).await + pub fn rfc8181(&self, publisher: PublisherHandle, msg_bytes: Bytes) -> KrillResult { + self.get_embedded()?.rfc8181(publisher, msg_bytes) } } @@ -596,7 +591,7 @@ impl KrillServer { let contact = match update { RepositoryUpdate::Embedded => { // Add to embedded publication server if not present - if self.get_embedded()?.get_publisher_details(&handle).await.is_err() { + if self.get_embedded()?.get_publisher_details(&handle).is_err() { let id_cert = { let ca = self.caserver.get_ca(&handle).await?; ca.id_cert().clone() @@ -604,10 +599,10 @@ impl KrillServer { // Add publisher let req = rfc8183::PublisherRequest::new(None, handle.clone(), id_cert); - self.add_publisher(req).await?; + self.add_publisher(req)?; } - RepositoryContact::embedded(self.get_embedded()?.repo_info_for(&handle).await?) + RepositoryContact::embedded(self.get_embedded()?.repo_info_for(&handle)?) } RepositoryUpdate::Rfc8181(response) => { // first check that the new repo can be contacted @@ -701,13 +696,13 @@ impl KrillServer { impl KrillServer { /// Handles a publish delta request sent to the API, or.. through /// the CmsProxy. - pub async fn handle_delta(&self, publisher: PublisherHandle, delta: PublishDelta) -> KrillEmptyResult { - self.get_embedded()?.publish(publisher, delta).await + pub fn handle_delta(&self, publisher: PublisherHandle, delta: PublishDelta) -> KrillEmptyResult { + self.get_embedded()?.publish(publisher, delta) } /// Handles a list request sent to the API, or.. through the CmsProxy. - pub async fn handle_list(&self, publisher: &PublisherHandle) -> KrillResult { - self.get_embedded()?.list(publisher).await + pub fn handle_list(&self, publisher: &PublisherHandle) -> KrillResult { + self.get_embedded()?.list(publisher) } } diff --git a/src/daemon/scheduler.rs b/src/daemon/scheduler.rs index 3266d1c8..2f06c8f8 100644 --- a/src/daemon/scheduler.rs +++ b/src/daemon/scheduler.rs @@ -36,6 +36,10 @@ pub struct Scheduler { /// Responsible for refreshing announcement information #[allow(dead_code)] // just need to keep this in scope announcements_refresh_sh: ScheduleHandle, + + /// Responsible for archiving old commands + #[allow(dead_code)] // just need to keep this in scope + archive_old_commands_sh: ScheduleHandle, } impl Scheduler { @@ -46,16 +50,17 @@ impl Scheduler { bgp_analyser: Arc, ca_refresh_rate: u32, ) -> Self { - let event_sh = make_event_sh(event_queue, caserver.clone(), pubserver); + let event_sh = make_event_sh(event_queue, caserver.clone(), pubserver.clone()); let republish_sh = make_republish_sh(caserver.clone()); - let ca_refresh_sh = make_ca_refresh_sh(caserver, ca_refresh_rate); + let ca_refresh_sh = make_ca_refresh_sh(caserver.clone(), ca_refresh_rate); let announcements_refresh_sh = make_announcements_refresh_sh(bgp_analyser); - + let archive_old_commands_sh = make_archive_old_commands_sh(caserver, pubserver); Scheduler { event_sh, republish_sh, ca_refresh_sh, announcements_refresh_sh, + archive_old_commands_sh, } } } @@ -230,3 +235,24 @@ fn make_announcements_refresh_sh(bgp_analyser: Arc) -> ScheduleHand }); scheduler.watch_thread(Duration::from_millis(100)) } + +fn make_archive_old_commands_sh(caserver: Arc, pubserver: Option>) -> ScheduleHandle { + let mut scheduler = clokwerk::Scheduler::new(); + scheduler.every(60.seconds()).run(move || { + let mut rt = Runtime::new().unwrap(); + rt.block_on(async { + if let Some(days) = CONFIG.archive_threshold_days { + if let Err(e) = caserver.archive_ca_commands(days).await { + error!("Failed to archive old CA commands: {}", e) + } + + if let Some(pubserver) = pubserver.as_ref() { + if let Err(e) = pubserver.history_archive_old(days) { + error!("Failed to archive old Publication Server commands: {}", e) + } + } + } + }) + }); + scheduler.watch_thread(Duration::from_millis(100)) +} diff --git a/src/pubd/pubserver.rs b/src/pubd/pubserver.rs index 2329b812..e8ec5794 100644 --- a/src/pubd/pubserver.rs +++ b/src/pubd/pubserver.rs @@ -42,7 +42,7 @@ pub struct PubServer { /// # Constructing /// impl PubServer { - pub async fn remove_if_empty( + pub fn remove_if_empty( rsync_base: &uri::Rsync, rrdp_base_uri: uri::Https, // for the RRDP files work_dir: &PathBuf, // for the aggregate stores @@ -52,8 +52,8 @@ impl PubServer { let mut pub_server_dir = work_dir.clone(); pub_server_dir.push(PUBSERVER_DIR); if pub_server_dir.exists() { - let server = PubServer::build(rsync_base, rrdp_base_uri, work_dir, rfc8181_log_dir, signer).await?; - if server.publishers().await?.is_empty() { + let server = PubServer::build(rsync_base, rrdp_base_uri, work_dir, rfc8181_log_dir, signer)?; + if server.publishers()?.is_empty() { let _result = fs::remove_dir_all(pub_server_dir); Ok(None) } else { @@ -64,7 +64,7 @@ impl PubServer { } } - pub async fn build( + pub fn build( rsync_base: &uri::Rsync, rrdp_base_uri: uri::Https, // for the RRDP files work_dir: &PathBuf, // for the aggregate stores @@ -105,7 +105,7 @@ impl PubServer { Handle::from_str(PUBSERVER_DFLT).unwrap() } - async fn repository(&self) -> KrillResult> { + fn repository(&self) -> KrillResult> { let handle = Self::repository_handle(); match self.store.get_latest(&handle) { @@ -118,8 +118,8 @@ impl PubServer { } /// Handle an RFC8181 request and sign the response - pub async fn rfc8181(&self, publisher_handle: PublisherHandle, msg_bytes: Bytes) -> KrillResult { - let repository = self.repository().await?; + pub fn rfc8181(&self, publisher_handle: PublisherHandle, msg_bytes: Bytes) -> KrillResult { + let repository = self.repository()?; let publisher = repository.get_publisher(&publisher_handle)?; let msg = ProtocolCms::decode(msg_bytes.clone(), false).map_err(|e| Error::Rfc8181Decode(e.to_string()))?; @@ -135,7 +135,7 @@ impl PubServer { let list_reply = publisher.list_current(); (rfc8181::Message::list_reply(list_reply), false) } - rfc8181::QueryMessage::PublishDelta(delta) => match self.publish(publisher_handle, delta).await { + rfc8181::QueryMessage::PublishDelta(delta) => match self.publish(publisher_handle, delta) { Ok(()) => (rfc8181::Message::success_reply(), true), Err(e) => { let error_code = e.to_rfc8181_error_code(); @@ -161,26 +161,26 @@ impl PubServer { } /// Let a known publisher publish in a repository. - pub async fn publish(&self, publisher: PublisherHandle, delta: PublishDelta) -> KrillResult<()> { + pub fn publish(&self, publisher: PublisherHandle, delta: PublishDelta) -> KrillResult<()> { let repository_handle = Self::repository_handle(); let cmd = CmdDet::publish(&repository_handle, publisher, delta); self.store.command(cmd)?; - self.write_repository().await + self.write_repository() } - pub async fn repo_stats(&self) -> KrillResult { - let repo = self.repository().await?; + pub fn repo_stats(&self) -> KrillResult { + let repo = self.repository()?; Ok(repo.stats().clone()) } - pub async fn publishers(&self) -> KrillResult> { - let repository = self.repository().await?; + pub fn publishers(&self) -> KrillResult> { + let repository = self.repository()?; Ok(repository.publishers()) } /// Returns a list reply for a known publisher in a repository - pub async fn list(&self, publisher: &PublisherHandle) -> KrillResult { - let repository = self.repository().await?; + pub fn list(&self, publisher: &PublisherHandle) -> KrillResult { + let repository = self.repository()?; let publisher = repository.get_publisher(publisher)?; Ok(publisher.list_current()) } @@ -189,31 +189,31 @@ impl PubServer { /// # Manage publishers /// impl PubServer { - pub async fn repo_info_for(&self, publisher: &PublisherHandle) -> KrillResult { - let repository = self.repository().await?; + pub fn repo_info_for(&self, publisher: &PublisherHandle) -> KrillResult { + let repository = self.repository()?; Ok(repository.repo_info_for(publisher)) } - pub async fn get_publisher_details(&self, publisher_handle: &PublisherHandle) -> KrillResult { - let repository = self.repository().await?; + pub fn get_publisher_details(&self, publisher_handle: &PublisherHandle) -> KrillResult { + let repository = self.repository()?; repository .get_publisher(publisher_handle) .map(|p| p.as_api_details(publisher_handle)) } /// Returns the RFC8183 Repository Response for the publisher - pub async fn repository_response( + pub fn repository_response( &self, rfc8181_uri: uri::Https, publisher: &PublisherHandle, ) -> KrillResult { - let repository = self.repository().await?; + let repository = self.repository()?; repository.repository_response(rfc8181_uri, publisher) } /// Adds a publisher. Will complain if a publisher already exists for this /// handle. Will also verify that the base_uri is allowed. - pub async fn create_publisher(&self, req: rfc8183::PublisherRequest) -> KrillResult<()> { + pub fn create_publisher(&self, req: rfc8183::PublisherRequest) -> KrillResult<()> { let repository_handle = Self::repository_handle(); let cmd = CmdDet::add_publisher(&repository_handle, req); self.store.command(cmd)?; @@ -224,11 +224,11 @@ impl PubServer { /// re-activation in future. Reason is that we never forget the history /// of the old publisher, and if handles are re-used by different /// entities that would get confusing. - pub async fn remove_publisher(&self, publisher: PublisherHandle) -> KrillResult<()> { + pub fn remove_publisher(&self, publisher: PublisherHandle) -> KrillResult<()> { let repository_handle = Self::repository_handle(); let cmd = CmdDet::remove_publisher(&repository_handle, publisher); self.store.command(cmd)?; - self.write_repository().await + self.write_repository() } } @@ -236,12 +236,22 @@ impl PubServer { /// impl PubServer { /// Update the RRDP files and rsync content on disk. - pub async fn write_repository(&self) -> KrillResult<()> { - let repository = self.repository().await?; + pub fn write_repository(&self) -> KrillResult<()> { + let repository = self.repository()?; repository.write() } } +/// # Manage history +/// +impl PubServer { + /// Archive old commands + pub fn history_archive_old(&self, days: i64) -> KrillResult<()> { + let handle = Self::repository_handle(); + self.store.archive_old_commands(&handle, days)?; + Ok(()) + } +} //------------ Tests --------------------------------------------------------- #[cfg(test)] @@ -285,28 +295,26 @@ mod tests { rfc8183::PublisherRequest::new(None, handle, id_cert.clone()) } - async fn make_server(work_dir: &PathBuf) -> PubServer { + fn make_server(work_dir: &PathBuf) -> PubServer { let signer = KrillSigner::build(work_dir).unwrap(); let signer = Arc::new(signer); - PubServer::build(&server_base_uri(), server_base_http_uri(), work_dir, None, signer) - .await - .unwrap() + PubServer::build(&server_base_uri(), server_base_http_uri(), work_dir, None, signer).unwrap() } - #[tokio::test] - async fn should_add_publisher() { + #[test] + fn should_add_publisher() { let d = test::tmp_dir(); - let server = make_server(&d).await; + let server = make_server(&d); let alice = publisher_alice(&d); let alice_handle = Handle::from_str("alice").unwrap(); let publisher_req = make_publisher_req(alice_handle.as_str(), alice.id_cert()); - server.create_publisher(publisher_req).await.unwrap(); + server.create_publisher(publisher_req).unwrap(); - let alice_found = server.get_publisher_details(&alice_handle).await.unwrap(); + let alice_found = server.get_publisher_details(&alice_handle).unwrap(); assert_eq!(alice_found.base_uri(), alice.base_uri()); assert_eq!(alice_found.id_cert(), alice.id_cert()); @@ -315,47 +323,47 @@ mod tests { let _ = fs::remove_dir_all(d); } - #[tokio::test] - async fn should_not_add_publisher_twice() { + #[test] + fn should_not_add_publisher_twice() { let d = test::tmp_dir(); - let server = make_server(&d).await; + let server = make_server(&d); let alice = publisher_alice(&d); let alice_handle = Handle::from_str("alice").unwrap(); let publisher_req = make_publisher_req(alice_handle.as_str(), alice.id_cert()); - server.create_publisher(publisher_req.clone()).await.unwrap(); + server.create_publisher(publisher_req.clone()).unwrap(); - match server.create_publisher(publisher_req).await { + match server.create_publisher(publisher_req) { Err(Error::PublisherDuplicate(name)) => assert_eq!(name, alice_handle), _ => panic!("Expected error"), } let _ = fs::remove_dir_all(d); } - #[tokio::test] - async fn should_list_files() { + #[test] + fn should_list_files() { let d = test::tmp_dir(); - let server = make_server(&d).await; + let server = make_server(&d); let alice = publisher_alice(&d); let alice_handle = Handle::from_str("alice").unwrap(); let publisher_req = make_publisher_req(alice_handle.as_str(), alice.id_cert()); - server.create_publisher(publisher_req).await.unwrap(); + server.create_publisher(publisher_req).unwrap(); - let list_reply = server.list(&alice_handle).await.unwrap(); + let list_reply = server.list(&alice_handle).unwrap(); assert_eq!(0, list_reply.elements().len()); let _ = fs::remove_dir_all(d); } - #[tokio::test] - async fn should_publish_files() { + #[test] + fn should_publish_files() { let d = test::tmp_dir(); - let server = make_server(&d).await; + let server = make_server(&d); // set up server with default repository, and publisher alice let alice = publisher_alice(&d); @@ -363,7 +371,7 @@ mod tests { let alice_handle = Handle::from_str("alice").unwrap(); let publisher_req = make_publisher_req(alice_handle.as_str(), alice.id_cert()); - server.create_publisher(publisher_req).await.unwrap(); + server.create_publisher(publisher_req).unwrap(); // get the file out of a list_reply fn find_in_reply<'a>(reply: &'a ListReply, uri: &uri::Rsync) -> Option<&'a ListElement> { @@ -386,10 +394,10 @@ mod tests { builder.add_publish(file2.as_publish()); let delta = builder.finish(); - server.publish(alice_handle.clone(), delta).await.unwrap(); + server.publish(alice_handle.clone(), delta).unwrap(); // Two files should now appear in the list - let list_reply = server.list(&alice_handle).await.unwrap(); + let list_reply = server.list(&alice_handle).unwrap(); assert_eq!(2, list_reply.elements().len()); assert!(find_in_reply(&list_reply, &test::rsync("rsync://localhost/repo/alice/file.txt")).is_some()); assert!(find_in_reply(&list_reply, &test::rsync("rsync://localhost/repo/alice/file2.txt")).is_some()); @@ -415,10 +423,10 @@ mod tests { builder.add_publish(file3.as_publish()); let delta = builder.finish(); - server.publish(alice_handle.clone(), delta).await.unwrap(); + server.publish(alice_handle.clone(), delta).unwrap(); // Two files should now appear in the list - let list_reply = server.list(&alice_handle).await.unwrap(); + let list_reply = server.list(&alice_handle).unwrap(); assert_eq!(2, list_reply.elements().len()); assert!(find_in_reply(&list_reply, &test::rsync("rsync://localhost/repo/alice/file.txt")).is_some()); @@ -439,7 +447,7 @@ mod tests { builder.add_publish(file_outside.as_publish()); let delta = builder.finish(); - match server.publish(alice_handle.clone(), delta).await { + match server.publish(alice_handle.clone(), delta) { Err(Error::Rfc8181Delta(PublicationDeltaError::UriOutsideJail(_, _))) => {} // ok _ => panic!("Expected error publishing outside of base uri jail"), } @@ -453,7 +461,7 @@ mod tests { builder.add_update(file2_update.as_update(file2.hash())); let delta = builder.finish(); - match server.publish(alice_handle.clone(), delta).await { + match server.publish(alice_handle.clone(), delta) { Err(Error::Rfc8181Delta(PublicationDeltaError::NoObjectForHashAndOrUri(_))) => {} _ => panic!("Expected error when file for update can't be found"), } @@ -463,7 +471,7 @@ mod tests { builder.add_withdraw(file2.as_withdraw()); let delta = builder.finish(); - match server.publish(alice_handle.clone(), delta).await { + match server.publish(alice_handle.clone(), delta) { Err(Error::Rfc8181Delta(PublicationDeltaError::NoObjectForHashAndOrUri(_))) => {} // ok _ => panic!("Expected error withdrawing file that does not exist"), } @@ -473,7 +481,7 @@ mod tests { builder.add_publish(file3.as_publish()); let delta = builder.finish(); - match server.publish(alice_handle, delta).await { + match server.publish(alice_handle, delta) { Err(Error::Rfc8181Delta(PublicationDeltaError::ObjectAlreadyPresent(uri))) => { assert_eq!(uri, test::rsync("rsync://localhost/repo/alice/file3.txt")) } diff --git a/src/publish/mod.rs b/src/publish/mod.rs index c4ca27b0..804ff069 100644 --- a/src/publish/mod.rs +++ b/src/publish/mod.rs @@ -41,7 +41,7 @@ impl CaPublisher { }; let list_reply = match &repo_contact { - RepositoryContact::Embedded(_) => self.get_embedded()?.list(ca_handle).await?, + RepositoryContact::Embedded(_) => self.get_embedded()?.list(ca_handle)?, RepositoryContact::Rfc8181(repo) => self.caserver.send_rfc8181_list(ca_handle, repo, false).await?, }; @@ -72,7 +72,7 @@ impl CaPublisher { }; match &repo_contact { - RepositoryContact::Embedded(_) => self.get_embedded()?.publish(ca_handle.clone(), delta).await?, + RepositoryContact::Embedded(_) => self.get_embedded()?.publish(ca_handle.clone(), delta)?, RepositoryContact::Rfc8181(repo) => self.caserver.send_rfc8181_delta(ca_handle, repo, delta, false).await?, }; @@ -90,14 +90,14 @@ impl CaPublisher { info!("Will perform best effort clean up of old repository: {}", repo); let list_reply = match repo { - RepositoryContact::Embedded(_) => self.get_embedded()?.list(ca_handle).await?, + RepositoryContact::Embedded(_) => self.get_embedded()?.list(ca_handle)?, RepositoryContact::Rfc8181(repo) => self.caserver.send_rfc8181_list(ca_handle, repo, true).await?, }; let delta = list_reply.into_withdraw_delta(); match repo { - RepositoryContact::Embedded(_) => self.get_embedded()?.publish(ca_handle.clone(), delta).await?, + RepositoryContact::Embedded(_) => self.get_embedded()?.publish(ca_handle.clone(), delta)?, RepositoryContact::Rfc8181(res) => self.caserver.send_rfc8181_delta(ca_handle, res, delta, true).await?, } diff --git a/test-resources/krill-init.conf b/test-resources/krill-init.conf index f2b43d44..caecd2da 100644 --- a/test-resources/krill-init.conf +++ b/test-resources/krill-init.conf @@ -22,6 +22,34 @@ # data_dir = "/var/lib/krill/data/" +# Archive publication events after X days. If you do NOT set this +# value then Krill will not do any archiving. +# +# If enabled this option will make sure that the republish events, +# where your CA simply generates a new Manifest and CRL are archived +# after the given number of days. +# +# If you run Krill as a Publication Server then this option will +# enable the archiving of publication deltas received by your server +# from CAs. +# +# Archived commands (containing e.g. details of when the change happened), +# and following events will be moved to an "archived" subdirectory under +# your data directory as follows: +# $data_dir/pubd/0/archived <-- for Publication Server +# $data_dir/cas/ca/archived <-- for a CA named 'ca' +# +# If you want to save space you can delete the files from these archived +# directories, e.g. from cron. However, you could also archive them in +# a different way: e.g. compress and move to long term storage. Krill will +# no longer need this data, but if you ever wanted to see these details in +# history you would need to move them back into the parent directory of +# the 'archived' directory. +# +# To enable this set the following key value pair. +# +### archive_threshold_days = 7 + # Specify the path to the PID file for Krill. # # Defaults to "krill.pid" under the 'data_dir' specified above.