Kvstore tx

.gitignore

@@ -2,7 +2,7 @@
 .idea/
 .vscode/
 .DS_Store
-data*
+data/
 krill.log
 target/
 tmp

Cargo.toml

@@ -36,8 +36,8 @@ hyper                 = { version = "^0.14", features = ["server"] }
 intervaltree          = "0.2.6"
 jmespatch             = { version = "^0.3", features = ["sync"], optional = true }
 kmip                  = { version = "0.4.2", package = "kmip-protocol", features = ["tls-with-openssl"], optional = true }
-kvx                   = { version = "0.6.0", features = ["macros"] }
-# kvx                   = { version = "0.6.0", git = "https://github.com/nlnetlabs/kvx", features = ["macros"] }
+# kvx                   = { version = "0.6.0", features = ["macros"] }
+kvx                   = { version = "0.7.0", git = "https://github.com/nlnetlabs/kvx", branch = "support-namespace-migrations", features = ["macros"] }
 libflate              = "^1"
 log                   = "^0.4"
 once_cell             = { version = "^1.7.2", optional = true }

src/bin/krillup.rs

@@ -1,82 +1,145 @@
 extern crate krill;
 
-use clap::{App, Arg};
-use log::info;
+use clap::{App, Arg, ArgMatches, SubCommand};
+use log::{info, LevelFilter};
 
 use krill::{
     constants::{KRILL_DEFAULT_CONFIG_FILE, KRILL_UP_APP, KRILL_VERSION},
-    daemon::{config::Config, properties::PropertiesManager},
-    upgrades::{prepare_upgrade_data_migrations, UpgradeMode},
+    daemon::{
+        config::{Config, LogType},
+        properties::PropertiesManager,
+    },
+    upgrades::{data_migration::migrate, prepare_upgrade_data_migrations, UpgradeMode},
 };
+use url::Url;
 
-#[tokio::main]
-async fn main() {
-    let matches = App::new(KRILL_UP_APP)
-        .version(KRILL_VERSION)
-        .about("\nThis tool can be used to reduce the risk and time needed for Krill upgrades, by preparing and verifying any data migrations that would be needed. The data_dir setting from the provided configuration file is used to find the data to migrate, and prepared data will be saved under 'data_dir/upgrade-data'.")
-        .arg(
-            Arg::with_name("config")
-                .short("c")
-                .long("config")
-                .value_name("FILE")
-                .help(&format!(
-                    "Override the path to the config file (default: '{}')",
-                    KRILL_DEFAULT_CONFIG_FILE
-                ))
-                .required(false),
-        )
-        .get_matches();
-
-    let config_file = matches.value_of("config").unwrap_or(KRILL_DEFAULT_CONFIG_FILE);
-
-    match Config::create(config_file, true) {
-        Ok(config) => {
-            let properties_manager = match PropertiesManager::create(&config.storage_uri, config.use_history_cache)
-            {
-                Ok(mgr) => mgr,
-                Err(e) => {
-                    eprintln!("*** Error Preparing Data Migration ***");
-                    eprintln!("{}", e);
-                    eprintln!();
-                    eprintln!("Note that your server data has NOT been modified. Do not upgrade krill itself yet!");
-                    eprintln!("If you did upgrade krill, then downgrade it to your previous installed version.");
-                    ::std::process::exit(1);
-                }
-            };
+fn main() {
+    let matches = make_matches();
+
+    match parse_matches(matches) {
+        Err(e) => {
+            eprintln!("{}", e);
+            ::std::process::exit(1);
+        }
+        Ok(mode) => match mode {
+            KrillUpMode::Prepare { config } => {
+                let properties_manager = match PropertiesManager::create(&config.storage_uri, config.use_history_cache)
+                {
+                    Ok(mgr) => mgr,
+                    Err(e) => {
+                        eprintln!("*** Error Preparing Data Migration ***");
+                        eprintln!("{}", e);
+                        eprintln!();
+                        eprintln!("Note that your server data has NOT been modified. Do not upgrade krill itself yet!");
+                        eprintln!("If you did upgrade krill, then downgrade it to your previous installed version.");
+                        ::std::process::exit(1);
+                    }
+                };
 
-            match prepare_upgrade_data_migrations(UpgradeMode::PrepareOnly, &config, &properties_manager) {
-                Err(e) => {
-                    eprintln!("*** Error Preparing Data Migration ***");
+                match prepare_upgrade_data_migrations(UpgradeMode::PrepareOnly, &config, &properties_manager) {
+                    Err(e) => {
+                        eprintln!("*** Error Preparing Data Migration ***");
+                        eprintln!("{}", e);
+                        eprintln!();
+                        eprintln!("Note that your server data has NOT been modified. Do not upgrade krill itself yet!");
+                        eprintln!("If you did upgrade krill, then downgrade it to your previous installed version.");
+                        ::std::process::exit(1);
+                    }
+                    Ok(opt) => match opt {
+                        None => {
+                            info!("No update needed");
+                        }
+                        Some(report) => {
+                            let from = report.versions().from();
+                            let to = report.versions().to();
+                            if report.data_migration() {
+                                info!("Prepared and verified upgrade from {} to {}.", from, to,);
+                            } else {
+                                info!("No preparation is needed for the upgrade from {} to {}.", from, to)
+                            }
+                        }
+                    },
+                }
+            }
+            KrillUpMode::Migrate { config, target } => {
+                if let Err(e) = migrate(config, target) {
+                    eprintln!("*** Error Migrating DATA ***");
                     eprintln!("{}", e);
                     eprintln!();
-                    eprintln!("Note that your server data has NOT been modified. Do not upgrade krill itself yet!");
-                    eprintln!("If you did upgrade krill, then downgrade it to your previous installed version.");
+                    eprintln!("Note that your server data has NOT been modified.");
                     ::std::process::exit(1);
                 }
-                Ok(opt) => match opt {
-                    None => {
-                        info!("No update needed");
-                    }
-                    Some(report) => {
-                        let from = report.versions().from();
-                        let to = report.versions().to();
-                        if report.data_migration() {
-                            info!(
-                                "Prepared and verified upgrade from {} to {}. Prepared data was saved to: {}",
-                                from,
-                                to,
-                                config.upgrade_storage_uri()
-                            );
-                        } else {
-                            info!("No preparation is needed for the upgrade from {} to {}.", from, to)
-                        }
-                    }
-                },
             }
-        }
-        Err(e) => {
-            eprintln!("Could not parse config: {}", e);
-            ::std::process::exit(1);
-        }
+        },
+    }
+}
+
+fn make_matches<'a>() -> ArgMatches<'a> {
+    let mut app = App::new(KRILL_UP_APP).version(KRILL_VERSION).about("\nThis tool can be used to reduce the risk and time needed for Krill upgrades, by preparing and verifying any data migrations that would be needed. The data_dir setting from the provided configuration file is used to find the data to migrate, and prepared data will be saved under 'data_dir/upgrade-data'.");
+
+    let mut prep_sub = SubCommand::with_name("prepare")
+        .about("Prepares a Krill upgrade data migration if needed by the new Krill version. This operation leaves the current data unmodified. You can run this operation while Krill is running. This tool will exit and report an error in case of any issues. To finish the migration, restart Krill. The migration process will continue to ensure it includes any changes after the preparation.");
+    prep_sub = add_config_arg(prep_sub);
+    app = app.subcommand(prep_sub);
+
+    let mut migrate_sub = SubCommand::with_name("migrate")
+        .about("Migrate Krill data to a different storage. Stop Krill before running this tool to ensure data does not change during migration. The original data in the storage defined in the config file is not modified. If your current data is for an older version of Krill, this tool will attempt to upgrade it. After successful migration, you can reconfigure Krill to use the new data storage and restart it.");
+    migrate_sub = add_config_arg(migrate_sub);
+    migrate_sub = add_new_storage_arg(migrate_sub);
+    app = app.subcommand(migrate_sub);
+
+    app.get_matches()
+}
+
+fn add_config_arg<'a, 'b>(app: App<'a, 'b>) -> App<'a, 'b> {
+    app.arg(
+        Arg::with_name("config")
+            .short("c")
+            .long("config")
+            .value_name("FILE")
+            .help("Override the path to the config file (default: '/etc/krill.conf').")
+            .required(false),
+    )
+}
+
+fn add_new_storage_arg<'a, 'b>(app: App<'a, 'b>) -> App<'a, 'b> {
+    app.arg(
+        Arg::with_name("target")
+            .short("t")
+            .long("target")
+            .value_name("URL")
+            .help("Provide the target storage URI string. E.g. local:///var/lib/krill or postgres://postgres@localhost/postgres.")
+            .required(true),
+    )
+}
+
+fn parse_matches(matches: ArgMatches) -> Result<KrillUpMode, String> {
+    if let Some(m) = matches.subcommand_matches("prepare") {
+        let config = parse_config(m)?;
+        Ok(KrillUpMode::Prepare { config })
+    } else if let Some(m) = matches.subcommand_matches("migrate") {
+        let target_str = m.value_of("target").ok_or("--target missing".to_string())?;
+        let target = Url::parse(target_str).map_err(|e| format!("cannot parse url: {}. Error: {}", target_str, e))?;
+
+        let config = parse_config(m)?;
+        Ok(KrillUpMode::Migrate { config, target })
+    } else {
+        Err("Cannot parse arguments. Use --help.".to_string())
     }
 }
+
+fn parse_config(m: &ArgMatches) -> Result<Config, String> {
+    let config_file = m.value_of("config").unwrap_or(KRILL_DEFAULT_CONFIG_FILE);
+    let mut config = Config::create(config_file, true)
+        .map_err(|e| format!("Cannot parse config file '{}'. Error: {}", config_file, e))?;
+
+    config.log_level = LevelFilter::Info;
+    config.log_type = LogType::Stderr;
+
+    Ok(config)
+}
+
+enum KrillUpMode {
+    Prepare { config: Config },
+    Migrate { config: Config, target: Url },
+}

src/cli/ta_client.rs

@@ -12,6 +12,7 @@ use std::{
 
 use bytes::Bytes;
 use clap::{App, Arg, ArgMatches, SubCommand};
+
 use log::LevelFilter;
 use rpki::{
     ca::idexchange::{self, ChildHandle, RepoInfo, ServiceUri},
@@ -28,7 +29,7 @@ use crate::{
         api::{AddChildRequest, ApiRepositoryContact, CertAuthInfo, IdCertInfo, RepositoryContact, Token},
         crypto::{KrillSigner, KrillSignerBuilder, OpenSslSignerConfig},
         error::Error as KrillError,
-        eventsourcing::{segment, AggregateStore, AggregateStoreError, Segment},
+        eventsourcing::{namespace, AggregateStore, AggregateStoreError, Namespace},
         util::{file, httpclient},
     },
     constants::{
@@ -1002,7 +1003,7 @@ struct TrustAnchorSignerManager {
 
 impl TrustAnchorSignerManager {
     fn create(config: Config) -> Result<Self, Error> {
-        let store = AggregateStore::create(&config.storage_uri, segment!("signer"), config.use_history_cache)
+        let store = AggregateStore::create(&config.storage_uri, namespace!("signer"), config.use_history_cache)
             .map_err(KrillError::AggregateStoreError)?;
         let ta_handle = TrustAnchorHandle::new("ta".into());
         let signer = config.signer()?;
@@ -1080,7 +1081,7 @@ impl TrustAnchorSignerManager {
 
     fn get_signer(&self) -> Result<Arc<TrustAnchorSigner>, Error> {
         if self.store.has(&self.ta_handle)? {
-            self.store.get_latest(&self.ta_handle).map_err(Error::StorageError)
+            self.store.get_latest(&self.ta_handle).map_err(Error::KrillError)
         } else {
             Err(Error::other("Trust Anchor Signer is not initialised."))
         }

src/commons/eventsourcing/cmd.rs

@@ -35,7 +35,7 @@ pub trait WithStorableDetails: Storable + Send + Sync {
 /// It should be storable in the same way as normal commands, sent to this
 /// aggregate type so that they can use the same kind of ProcessedCommand
 /// and CommandHistoryRecord
-pub trait InitCommand: fmt::Display + Send + Sync {
+pub trait InitCommand: Clone + fmt::Display + Send + Sync {
     /// Identify the type of storable component for this command. Commands
     /// may contain short-lived things (e.g. an Arc<Signer>) or even secrets
     /// which should not be persisted.
@@ -57,6 +57,7 @@ pub trait InitCommand: fmt::Display + Send + Sync {
 
 /// Convenience wrapper so that implementations can just implement
 /// ['InitCommandDetails'] and leave the id and version boilerplate.
+#[derive(Clone, Debug)]
 pub struct SentInitCommand<I: InitCommandDetails> {
     handle: MyHandle,
     details: I,
@@ -103,7 +104,7 @@ impl<I: InitCommandDetails> fmt::Display for SentInitCommand<I> {
 
 /// Implement this for an enum with CommandDetails, so you you can reuse the
 /// id and version boilerplate from ['SentCommand'].
-pub trait InitCommandDetails: fmt::Display + Send + Sync + 'static {
+pub trait InitCommandDetails: Clone + fmt::Display + Send + Sync + 'static {
     type StorableDetails: WithStorableDetails;
 
     fn store(&self) -> Self::StorableDetails;
@@ -116,7 +117,7 @@ pub trait InitCommandDetails: fmt::Display + Send + Sync + 'static {
 /// Think of this as the data container for your update API, plus some
 /// meta-data to ensure that the command is sent to the right instance of an
 /// Aggregate, and that concurrency issues are handled.
-pub trait Command: fmt::Display + Send + Sync {
+pub trait Command: Clone + fmt::Display + Send + Sync {
     /// Identify the type of storable component for this command. Commands
     /// may contain short-lived things (e.g. an Arc<Signer>) or even secrets
     /// which should not be persisted.
@@ -211,7 +212,7 @@ impl<C: CommandDetails> fmt::Display for SentCommand<C> {
 
 /// Implement this for an enum with CommandDetails, so you you can reuse the
 /// id and version boilerplate from ['SentCommand'].
-pub trait CommandDetails: fmt::Display + Send + Sync + 'static {
+pub trait CommandDetails: Clone + fmt::Display + Send + Sync + 'static {
     type Event: Event;
     type StorableDetails: WithStorableDetails;