diff --git a/Cargo.lock b/Cargo.lock index 86d06a3f286f..af76d864c61d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -14973,6 +14973,34 @@ dependencies = [ "tempfile", ] +[[package]] +name = "ic-subnet-merging" +version = "0.9.0" +dependencies = [ + "anyhow", + "clap", + "futures", + "ic-base-types", + "ic-cup-explorer", + "ic-protobuf", + "ic-recovery", + "ic-registry-client-helpers", + "ic-registry-routing-table", + "ic-registry-subnet-type", + "ic-state-layout", + "ic-subnet-tools", + "ic-test-utilities-tmpdir", + "ic-types", + "reqwest", + "serde", + "serde_json", + "slog", + "strum 0.26.3", + "strum_macros 0.26.4", + "tokio", + "url", +] + [[package]] name = "ic-subnet-splitting" version = "0.9.0" @@ -14981,11 +15009,7 @@ dependencies = [ "candid", "clap", "csv", - "hex", - "ic-agent", "ic-base-types", - "ic-crypto-utils-threshold-sig", - "ic-crypto-utils-threshold-sig-der", "ic-management-canister-types-private", "ic-metrics", "ic-protobuf", @@ -14996,6 +15020,7 @@ dependencies = [ "ic-state-machine-tests", "ic-state-manager", "ic-state-tool", + "ic-subnet-tools", "ic-test-utilities-logger", "ic-test-utilities-tmpdir", "ic-test-utilities-types", @@ -15004,13 +15029,33 @@ dependencies = [ "ic-universal-canister", "proxy_canister", "serde", - "serde_cbor", "slog", "strum 0.26.3", "strum_macros 0.26.4", "url", ] +[[package]] +name = "ic-subnet-tools" +version = "0.9.0" +dependencies = [ + "hex", + "ic-agent", + "ic-base-types", + "ic-crypto-utils-threshold-sig", + "ic-crypto-utils-threshold-sig-der", + "ic-protobuf", + "ic-recovery", + "ic-registry-routing-table", + "ic-registry-subnet-type", + "ic-state-manager", + "ic-state-tool", + "ic-types", + "serde_cbor", + "slog", + "url", +] + [[package]] name = "ic-sys" version = "0.9.0" @@ -18230,20 +18275,29 @@ dependencies = [ "candid", "canister-test", "dfn_candid", + "hex", "ic-agent", "ic-base-types", "ic-management-canister-types 0.8.0", + "ic-nns-governance-api", + "ic-recovery", "ic-registry-subnet-type", + "ic-state-layout", + "ic-subnet-merging", "ic-system-test-driver", "ic-types", + "ic-universal-canister", "ic-utils 0.49.1", + "ic_consensus_system_test_utils", "itertools 0.12.1", "rand 0.8.6", "rand_chacha 0.3.1", + "registry-canister", "rejoin-test-lib", "slog", "tempfile", "tokio", + "url", "xnet-test", ] diff --git a/Cargo.toml b/Cargo.toml index ba21cb9ee93e..f497c666c00a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -341,7 +341,9 @@ members = [ "rs/protobuf/generator", "rs/query_stats", "rs/recovery", + "rs/recovery/subnet_merging", "rs/recovery/subnet_splitting", + "rs/recovery/subnet_tools", "rs/registry/admin", "rs/registry/admin-derive", "rs/registry/canister", diff --git a/rs/cup_explorer/src/lib.rs b/rs/cup_explorer/src/lib.rs index 295e5c133dee..dd7b8367f985 100644 --- a/rs/cup_explorer/src/lib.rs +++ b/rs/cup_explorer/src/lib.rs @@ -43,7 +43,7 @@ pub async fn get_catchup_content(url: &Url) -> Result } /// Fetches the CatchUp package, if it's present. -async fn get_cup(url: &Url) -> Result, String> { +pub async fn get_cup(url: &Url) -> Result, String> { let agent = Agent::new(url.clone(), Sender::Anonymous); agent .query_cup_endpoint(None) diff --git a/rs/recovery/src/app_subnet_recovery.rs b/rs/recovery/src/app_subnet_recovery.rs index a942bd4d1d55..3651f539d7a7 100644 --- a/rs/recovery/src/app_subnet_recovery.rs +++ b/rs/recovery/src/app_subnet_recovery.rs @@ -608,6 +608,7 @@ impl RecoveryIterator for AppSubnetRecovery { None, self.params.initial_dkg_subnet_id, self.params.chain_key_subnet_id, + /*time=*/ None, )?)) } diff --git a/rs/recovery/src/lib.rs b/rs/recovery/src/lib.rs index d0c54a619d02..95a247cf305b 100644 --- a/rs/recovery/src/lib.rs +++ b/rs/recovery/src/lib.rs @@ -817,6 +817,7 @@ impl Recovery { registry_params: Option, initial_dkg_subnet_id: Option, chain_key_subnet_id: Option, + time: Option, ) -> RecoveryResult> { let chain_key_config = chain_key_subnet_id .map(|id| match self.registry_helper.get_chain_key_config(id) { @@ -847,7 +848,11 @@ impl Recovery { chain_key_config, replacement_nodes, registry_params, - SystemTime::now(), + // The block time the recovered subnet starts from. Defaults + // to now, which is what a recovery replaying up to the + // present wants; a subnet merge passes the time it computed + // from the states it merged instead. + time.unwrap_or_else(SystemTime::now), ), }) } @@ -1209,7 +1214,7 @@ pub fn get_available_nodes_heights_from_metrics( } /// Lookup node IDs and corresponding IP addresses of all members of the given subnet -fn get_member_node_ids_and_ips( +pub fn get_member_node_ids_and_ips( registry_helper: &RegistryHelper, subnet_id: SubnetId, ) -> RecoveryResult> { diff --git a/rs/recovery/src/nns_recovery_failover_nodes.rs b/rs/recovery/src/nns_recovery_failover_nodes.rs index 11e5f0420340..b2f41bed9e2d 100644 --- a/rs/recovery/src/nns_recovery_failover_nodes.rs +++ b/rs/recovery/src/nns_recovery_failover_nodes.rs @@ -431,6 +431,7 @@ impl RecoveryIterator for NNSRecoveryFailoverNodes { Some(registry_params), None, None, + /*time=*/ None, )?)) } else { Err(RecoveryError::StepSkipped) diff --git a/rs/recovery/src/registry_helper.rs b/rs/recovery/src/registry_helper.rs index 20eee4a4f6b7..7ec5db028dce 100644 --- a/rs/recovery/src/registry_helper.rs +++ b/rs/recovery/src/registry_helper.rs @@ -102,6 +102,29 @@ impl RegistryHelper { }) } + /// Returns the subnet record of the given subnet as of `registry_version`, + /// rather than as of the latest version [Self::get_subnet_record] reads. + /// + /// Polls the [RegistryReplicator] first, as every other getter does: a + /// version that the local store has not caught up with yet is not readable, + /// and the caller may well be asking about one that was just created. + pub fn get_subnet_record_at_version( + &self, + subnet_id: SubnetId, + registry_version: RegistryVersion, + ) -> RecoveryResult> { + let _ = self.latest_registry_version()?; + + self.registry_client() + .get_subnet_record(subnet_id, registry_version) + .map_err(|err| { + RecoveryError::RegistryError(format!( + "Failed to get the record of subnet {subnet_id} at registry version \ + {registry_version}: {err}" + )) + }) + } + /// Returns the [SubnetRecord] of the given subnet. pub fn get_subnet_record(&self, subnet_id: SubnetId) -> VersionedRecoveryResult { self.get(|registry_version, registry_client| { diff --git a/rs/recovery/src/steps.rs b/rs/recovery/src/steps.rs index d582eafa9af7..8c5f288fbd93 100644 --- a/rs/recovery/src/steps.rs +++ b/rs/recovery/src/steps.rs @@ -274,7 +274,11 @@ impl Step for MergeCertificationPoolsStep { } } -pub(crate) struct DownloadIcDataStep { +/// Downloads data (the state, the consensus pool, ...) of a node into a +/// working directory. Public so that a tool that works with more than one +/// working directory, such as subnet merging, can direct the download at the +/// right one. +pub struct DownloadIcDataStep { pub logger: Logger, pub ssh_helper: SshHelper, pub backup_dir: PathBuf, diff --git a/rs/recovery/subnet_merging/BUILD.bazel b/rs/recovery/subnet_merging/BUILD.bazel new file mode 100644 index 000000000000..da493020eb70 --- /dev/null +++ b/rs/recovery/subnet_merging/BUILD.bazel @@ -0,0 +1,58 @@ +load("@rules_rust//rust:defs.bzl", "rust_binary", "rust_library", "rust_test") + +DEPENDENCIES = [ + # Keep sorted. + "//rs/cup_explorer", + "//rs/protobuf", + "//rs/recovery", + "//rs/recovery/subnet_tools", + "//rs/registry/helpers", + "//rs/registry/routing_table", + "//rs/registry/subnet_type", + "//rs/state_layout", + "//rs/types/base_types", + "//rs/types/types", + "@crate_index//:anyhow", + "@crate_index//:clap", + "@crate_index//:futures", + "@crate_index//:reqwest", + "@crate_index//:serde", + "@crate_index//:serde_json", + "@crate_index//:slog", + "@crate_index//:strum", + "@crate_index//:tokio", + "@crate_index//:url", +] + +rust_library( + name = "subnet_merging", + srcs = glob( + ["src/**/*.rs"], + exclude = ["src/main.rs"], + ), + crate_name = "ic_subnet_merging", + proc_macro_deps = ["@crate_index//:strum_macros"], + version = "0.1.0", + visibility = ["//rs:system-tests-pkg"], + deps = DEPENDENCIES, +) + +rust_binary( + name = "subnet-merging-tool", + srcs = ["src/main.rs"], + visibility = ["//rs:release-pkg"], + deps = DEPENDENCIES + [ + # Keep sorted. + ":subnet_merging", + "//rs/canister_sandbox:backend_lib", + ], +) + +rust_test( + name = "subnet_merging_tool_test", + crate = "subnet_merging", + deps = DEPENDENCIES + [ + # Keep sorted. + "//rs/test_utilities/tmpdir", + ], +) diff --git a/rs/recovery/subnet_merging/Cargo.toml b/rs/recovery/subnet_merging/Cargo.toml new file mode 100644 index 000000000000..21d7650501d5 --- /dev/null +++ b/rs/recovery/subnet_merging/Cargo.toml @@ -0,0 +1,38 @@ +[package] +name = "ic-subnet-merging" +version.workspace = true +authors.workspace = true +edition.workspace = true +description.workspace = true +documentation.workspace = true + +[dependencies] +anyhow = { workspace = true } +clap = { workspace = true } +futures = { workspace = true } +ic-base-types = { path = "../../types/base_types/" } +ic-cup-explorer = { path = "../../cup_explorer" } +ic-protobuf = { path = "../../protobuf" } +ic-recovery = { path = "../" } +ic-registry-client-helpers = { path = "../../registry/helpers" } +ic-registry-routing-table = { path = "../../registry/routing_table" } +ic-registry-subnet-type = { path = "../../registry/subnet_type" } +ic-state-layout = { path = "../../state_layout" } +ic-subnet-tools = { path = "../subnet_tools" } +ic-types = { path = "../../types/types" } +reqwest = { workspace = true } +serde = { workspace = true } +serde_json = { workspace = true } +slog = { workspace = true } + +strum = { workspace = true } +strum_macros = { workspace = true } +tokio = { workspace = true } +url = { workspace = true } + +[dev-dependencies] +ic-test-utilities-tmpdir = { path = "../../test_utilities/tmpdir" } + +[[bin]] +name = "subnet-merging-tool" +path = "src/main.rs" diff --git a/rs/recovery/subnet_merging/src/admin_helper.rs b/rs/recovery/subnet_merging/src/admin_helper.rs new file mode 100644 index 000000000000..1a592000b35f --- /dev/null +++ b/rs/recovery/subnet_merging/src/admin_helper.rs @@ -0,0 +1,183 @@ +use ic_base_types::SubnetId; +use ic_recovery::admin_helper::{ + AdminHelper, CommandHelper, IcAdmin, SSH_READONLY_ACCESS_ARG, SUMMARY_ARG, quote, +}; +use ic_subnet_tools::admin_helper::{DESTINATION_SUBNET_ARG, SOURCE_SUBNET_ARG, SUBNET_ARG}; + +const SUBNET_ID_ARG: &str = "subnet-id"; + +/// Propose to label the subnet as "cooling down", i.e. to have it stop +/// accepting ingress messages, answering queries and executing canister +/// messages, so that it quiesces and can be merged into another subnet. +/// +/// Optionally adds a ssh-readonly-access key to the subnet, which is needed to +/// download its state later on. +pub(crate) fn get_propose_to_cool_down_subnet_command( + admin_helper: &AdminHelper, + subnet_id: SubnetId, + key: &Option, +) -> IcAdmin { + let mut ic_admin = admin_helper.get_ic_admin_cmd_base(); + + ic_admin + .add_positional_argument("propose-to-update-subnet") + .add_argument(SUBNET_ARG, subnet_id) + .add_argument( + SUMMARY_ARG, + quote(format!( + "Label subnet {subnet_id} as cooling down and optionally update ssh readonly access", + )), + ) + .add_argument("cooling-down", true); + + if let Some(key) = key { + ic_admin.add_argument(SSH_READONLY_ACCESS_ARG, quote(key)); + } + + admin_helper.add_proposer_args(&mut ic_admin); + + ic_admin +} + +/// Propose to reroute the canister ID ranges of the source subnet to the +/// destination subnet, i.e. to merge the former into the latter. +pub(crate) fn get_propose_to_merge_subnets_command( + admin_helper: &AdminHelper, + source_subnet_id: SubnetId, + destination_subnet_id: SubnetId, +) -> IcAdmin { + let mut ic_admin = admin_helper.get_ic_admin_cmd_base(); + + ic_admin + .add_positional_argument("propose-to-merge-subnets") + .add_argument( + SUMMARY_ARG, + quote(format!( + "Merge subnet {source_subnet_id} into subnet {destination_subnet_id}", + )), + ) + .add_argument(SOURCE_SUBNET_ARG, source_subnet_id) + .add_argument(DESTINATION_SUBNET_ARG, destination_subnet_id); + + admin_helper.add_proposer_args(&mut ic_admin); + + ic_admin +} + +/// Propose to delete the subnet that was merged away and that hosts no canister +/// ID range anymore. +pub(crate) fn get_propose_to_delete_subnet_command( + admin_helper: &AdminHelper, + subnet_id: SubnetId, +) -> IcAdmin { + let mut ic_admin = admin_helper.get_ic_admin_cmd_base(); + + ic_admin + .add_positional_argument("propose-to-delete-subnet") + .add_argument( + SUMMARY_ARG, + quote(format!( + "Delete subnet {subnet_id}, which was merged into another subnet and hosts no \ + canister id range anymore", + )), + ) + .add_argument(SUBNET_ID_ARG, subnet_id); + + admin_helper.add_proposer_args(&mut ic_admin); + + ic_admin +} + +#[cfg(test)] +mod tests { + use super::*; + + use ic_base_types::PrincipalId; + use url::Url; + + use std::{path::PathBuf, str::FromStr}; + + const FAKE_IC_ADMIN: &str = "/fake/ic/admin/dir/ic-admin"; + const FAKE_NNS_URL: &str = "https://fake_nns_url.com:8080"; + const FAKE_SUBNET_ID_1: &str = + "gpvux-2ejnk-3hgmh-cegwf-iekfc-b7rzs-hrvep-5euo2-3ywz3-k3hcb-cqe"; + const FAKE_SUBNET_ID_2: &str = + "mklno-zzmhy-zutel-oujwg-dzcli-h6nfy-2serg-gnwru-vuwck-hcxit-wqe"; + const SSH_KEY: &str = "fake ssh key"; + + #[test] + fn get_propose_to_cool_down_subnet_command_test() { + let result = get_propose_to_cool_down_subnet_command( + &fake_admin_helper(), + subnet_id_from_str(FAKE_SUBNET_ID_1), + &Some(SSH_KEY.to_string()), + ) + .join(" "); + + assert_eq!( + result, + "/fake/ic/admin/dir/ic-admin \ + --nns-url \"https://fake_nns_url.com:8080/\" \ + propose-to-update-subnet \ + --subnet gpvux-2ejnk-3hgmh-cegwf-iekfc-b7rzs-hrvep-5euo2-3ywz3-k3hcb-cqe \ + --summary \"Label subnet gpvux-2ejnk-3hgmh-cegwf-iekfc-b7rzs-hrvep-5euo2-3ywz3-k3hcb-cqe as cooling down and optionally update ssh readonly access\" \ + --cooling-down true \ + --ssh-readonly-access \"fake ssh key\" \ + --test-neuron-proposer" + ); + } + + #[test] + fn get_propose_to_merge_subnets_command_test() { + let result = get_propose_to_merge_subnets_command( + &fake_admin_helper(), + subnet_id_from_str(FAKE_SUBNET_ID_1), + subnet_id_from_str(FAKE_SUBNET_ID_2), + ) + .join(" "); + + assert_eq!( + result, + "/fake/ic/admin/dir/ic-admin \ + --nns-url \"https://fake_nns_url.com:8080/\" \ + propose-to-merge-subnets \ + --summary \"Merge subnet gpvux-2ejnk-3hgmh-cegwf-iekfc-b7rzs-hrvep-5euo2-3ywz3-k3hcb-cqe into subnet mklno-zzmhy-zutel-oujwg-dzcli-h6nfy-2serg-gnwru-vuwck-hcxit-wqe\" \ + --source-subnet gpvux-2ejnk-3hgmh-cegwf-iekfc-b7rzs-hrvep-5euo2-3ywz3-k3hcb-cqe \ + --destination-subnet mklno-zzmhy-zutel-oujwg-dzcli-h6nfy-2serg-gnwru-vuwck-hcxit-wqe \ + --test-neuron-proposer" + ); + } + + #[test] + fn get_propose_to_delete_subnet_command_test() { + let result = get_propose_to_delete_subnet_command( + &fake_admin_helper(), + subnet_id_from_str(FAKE_SUBNET_ID_1), + ) + .join(" "); + + assert_eq!( + result, + "/fake/ic/admin/dir/ic-admin \ + --nns-url \"https://fake_nns_url.com:8080/\" \ + propose-to-delete-subnet \ + --summary \"Delete subnet gpvux-2ejnk-3hgmh-cegwf-iekfc-b7rzs-hrvep-5euo2-3ywz3-k3hcb-cqe, which was merged into another subnet and hosts no canister id range anymore\" \ + --subnet-id gpvux-2ejnk-3hgmh-cegwf-iekfc-b7rzs-hrvep-5euo2-3ywz3-k3hcb-cqe \ + --test-neuron-proposer" + ); + } + + fn fake_admin_helper() -> AdminHelper { + AdminHelper::new( + PathBuf::from(FAKE_IC_ADMIN), + Url::try_from(FAKE_NNS_URL).unwrap(), + /*neuron_args=*/ None, + ) + } + + fn subnet_id_from_str(subnet_id: &str) -> SubnetId { + PrincipalId::from_str(subnet_id) + .map(SubnetId::from) + .unwrap() + } +} diff --git a/rs/recovery/subnet_merging/src/layout.rs b/rs/recovery/subnet_merging/src/layout.rs new file mode 100644 index 000000000000..42fd5fcd5065 --- /dev/null +++ b/rs/recovery/subnet_merging/src/layout.rs @@ -0,0 +1,193 @@ +use crate::{ + target_subnet::TargetSubnet, + utils::{ + COOLING_DOWN_REGISTRY_VERSION_FILE, MERGE_REGISTRY_VERSION_FILE, read_registry_version, + write_registry_version, + }, +}; + +use ic_base_types::SubnetId; +use ic_recovery::{ + CHECKPOINTS, CUPS_DIR, IC_STATE, Recovery, error::RecoveryResult, file_sync_helper::create_dir, +}; +use ic_state_layout::StateLayout; +use ic_types::Height; + +use std::path::{Path, PathBuf}; + +/// The name the orchestrator stores the latest CUP of a subnet under. +pub(crate) const CUP_FILE_NAME: &str = "cup.types.v1.CatchUpPackage.pb"; + +#[derive(Clone)] +/// Describes the layout of the working directory of subnet merging: +/// +/// |-- root/ +/// | |-- ${destination_subnet_id}.manifest +/// | |-- ${source_subnet_id}.manifest +/// | |-- ${source_subnet_id}.pem +/// | |-- ${destination_subnet_id}.pem +/// | |-- merged.manifest +/// | |-- merged_state_params.json +/// | |-- cooling_down_registry_version +/// | |-- merge_registry_version +/// | |-- nns.pem +/// | |-- ${source_subnet_id}.pruned_state_tree.cbor +/// | |-- ${destination_subnet_id}.pruned_state_tree.cbor +/// | |-- ${source_subnet_id}.halting_cup.pb +/// | |-- ${destination_subnet_id}.halting_cup.pb +/// | |-- (destination_|merged_)work_dir/ +/// | | |-- data/ +/// | | | |-- cups/cup.types.v1.CatchUpPackage.pb +/// | | | |-- ic_state/states_metadata.pbuf +/// | | | |-- ic_state/checkpoints/ +/// | | | | |-- 1/ +pub(crate) struct Layout { + root: PathBuf, + source_working_dir: PathBuf, + + nns_public_key: PathBuf, + merged_state_manifest: PathBuf, + merged_state_params: PathBuf, + cooling_down_registry_version: PathBuf, + merge_registry_version: PathBuf, +} + +impl Layout { + pub(crate) fn new(recovery: &Recovery) -> Self { + Self { + root: recovery.recovery_dir.clone(), + source_working_dir: recovery.work_dir.clone(), + nns_public_key: recovery.recovery_dir.join("nns.pem"), + merged_state_manifest: recovery.recovery_dir.join("merged.manifest"), + merged_state_params: recovery.recovery_dir.join("merged_state_params.json"), + cooling_down_registry_version: recovery + .recovery_dir + .join(COOLING_DOWN_REGISTRY_VERSION_FILE), + merge_registry_version: recovery.recovery_dir.join(MERGE_REGISTRY_VERSION_FILE), + } + } + + /// Creates the working directories of the two subnets being merged and of + /// the merged state. `Recovery` only creates the one working directory a + /// recovery has, which is the source subnet's here. + pub(crate) fn create_dirs(&self) -> RecoveryResult<()> { + for target_subnet in [ + TargetSubnet::Source, + TargetSubnet::Destination, + TargetSubnet::Merged, + ] { + create_dir(&self.data_dir(target_subnet))?; + } + + Ok(()) + } + + pub(crate) fn nns_public_key_file(&self) -> &Path { + &self.nns_public_key + } + + pub(crate) fn merged_state_manifest_file(&self) -> &Path { + &self.merged_state_manifest + } + + pub(crate) fn merged_state_params_file(&self) -> &Path { + &self.merged_state_params + } + + /// The registry version at which the subnet that is merged away was + /// labeled "cooling down", i.e. the `V` of the merge readiness condition. + pub(crate) fn write_cooling_down_registry_version(&self, version: u64) -> RecoveryResult<()> { + write_registry_version(&self.cooling_down_registry_version, version) + } + + pub(crate) fn read_cooling_down_registry_version(&self) -> RecoveryResult { + read_registry_version(&self.cooling_down_registry_version) + } + + /// The registry version the merge was applied at, i.e. the one every other + /// subnet has to have reached before the merged subnet may be deleted. + pub(crate) fn write_merge_registry_version(&self, version: u64) -> RecoveryResult<()> { + write_registry_version(&self.merge_registry_version, version) + } + + pub(crate) fn read_merge_registry_version(&self) -> RecoveryResult { + read_registry_version(&self.merge_registry_version) + } + + pub(crate) fn actual_manifest_file(&self, subnet_id: SubnetId) -> PathBuf { + self.root.join(format!("{subnet_id}.manifest")) + } + + pub(crate) fn subnet_public_key_file(&self, subnet_id: SubnetId) -> PathBuf { + self.root.join(format!("{subnet_id}.pem")) + } + + pub(crate) fn pruned_state_tree_file(&self, subnet_id: SubnetId) -> PathBuf { + self.root + .join(format!("{subnet_id}.pruned_state_tree.cbor")) + } + + /// The copy of the CUP a subnet halted at that is pulled off its node while + /// waiting for the halt, before the state is downloaded. + pub(crate) fn halting_cup_file(&self, subnet_id: SubnetId) -> PathBuf { + self.root.join(format!("{subnet_id}.halting_cup.pb")) + } + + /// The CUP that came with the downloaded state, i.e. the one the validation + /// of the downloaded state checks against. + pub(crate) fn downloaded_cup_file(&self, target_subnet: TargetSubnet) -> PathBuf { + self.data_dir(target_subnet) + .join(CUPS_DIR) + .join(CUP_FILE_NAME) + } + + pub(crate) fn work_dir(&self, target_subnet: TargetSubnet) -> PathBuf { + match target_subnet { + TargetSubnet::Source => self.source_working_dir.clone(), + TargetSubnet::Destination => self.root.join("destination_working_dir"), + TargetSubnet::Merged => self.root.join("merged_working_dir"), + } + } + + /// Where the state of a subnet is downloaded to when the downloaded state + /// is to be kept, before it is copied into the working directory. + pub(crate) fn original_data_dir(&self, target_subnet: TargetSubnet) -> PathBuf { + self.work_dir(target_subnet).join("original_data") + } + + pub(crate) fn data_dir(&self, target_subnet: TargetSubnet) -> PathBuf { + self.work_dir(target_subnet).join("data") + } + + pub(crate) fn ic_state_dir(&self, target_subnet: TargetSubnet) -> PathBuf { + self.data_dir(target_subnet).join(IC_STATE) + } + + pub(crate) fn checkpoints_dir(&self, target_subnet: TargetSubnet) -> PathBuf { + self.ic_state_dir(target_subnet).join(CHECKPOINTS) + } + + pub(crate) fn checkpoint_dir(&self, target_subnet: TargetSubnet, height: Height) -> PathBuf { + self.checkpoints_dir(target_subnet) + .join(StateLayout::checkpoint_name(height)) + } + + pub(crate) fn latest_checkpoint_dir( + &self, + target_subnet: TargetSubnet, + ) -> RecoveryResult { + let checkpoints_dir = self.checkpoints_dir(target_subnet); + + let (max_name, _) = Recovery::get_latest_checkpoint_name_and_height(&checkpoints_dir)?; + + Ok(checkpoints_dir.join(max_name)) + } + + pub(crate) fn latest_checkpoint_height( + &self, + target_subnet: TargetSubnet, + ) -> RecoveryResult { + Recovery::get_latest_checkpoint_name_and_height(&self.checkpoints_dir(target_subnet)) + .map(|(_, height)| height) + } +} diff --git a/rs/recovery/subnet_merging/src/lib.rs b/rs/recovery/subnet_merging/src/lib.rs new file mode 100644 index 000000000000..ff5a666ff2cc --- /dev/null +++ b/rs/recovery/subnet_merging/src/lib.rs @@ -0,0 +1,9 @@ +pub mod readiness; +pub mod subnet_merging; +pub mod utils; + +mod admin_helper; +mod layout; +mod metrics_helper; +mod steps; +mod target_subnet; diff --git a/rs/recovery/subnet_merging/src/main.rs b/rs/recovery/subnet_merging/src/main.rs new file mode 100644 index 000000000000..233f4395357f --- /dev/null +++ b/rs/recovery/subnet_merging/src/main.rs @@ -0,0 +1,178 @@ +use anyhow::Context; +use clap::Parser; +use ic_base_types::SubnetId; +use ic_recovery::{NeuronArgs, RecoveryArgs, cli, util}; +use ic_subnet_merging::subnet_merging::{SubnetMerging, SubnetMergingArgs}; +use ic_subnet_tools::validation::validate_artifacts; +use ic_types::ReplicaVersion; +use slog::{Logger, info, warn}; +use url::Url; + +use std::path::PathBuf; + +const FORUM_ANNOUNCEMENT_TEMPLATE_URL: &str = + "https://wiki.internetcomputer.org/wiki/Subnet_merging_forum_announcement_template"; + +#[derive(Parser)] +struct MergeArgs { + #[clap( + short = 'r', + long, + alias = "registry-url", + default_value = "https://ic0.app" + )] + /// The URL of an NNS entry point. That is, the URL of any replica on the + /// NNS subnet. + nns_url: Url, + + /// replica version of ic-admin binary + #[clap(long)] + replica_version: Option, + + /// The directory to do the subnet merging in + #[clap(long)] + dir: PathBuf, + + /// The path to a private key to be considered for admin SSH connections + #[clap(long)] + admin_key_file: Option, + + /// Flag to enter test mode + #[clap(long)] + test: bool, + + /// Flag to make the tool non interactive. No input from the user is requested. + #[clap(long)] + pub skip_prompts: bool, + + #[clap(flatten)] + subnet_merging_args: SubnetMergingArgs, +} + +#[derive(Parser)] +struct ValidateArgs { + /// Path to the State Tree signed by the NNS + #[clap(long)] + state_tree_path: PathBuf, + + /// (Optional) path to the NNS public key. If not set, the built-in public key is used. + #[clap(long)] + nns_public_key_path: Option, + + /// Path to the CUP the subnet halted at, retrieved from one of its nodes. + #[clap(long)] + cup_path: PathBuf, + + /// Path to the manifest computed from the state the subnet halted at. + #[clap(long)] + state_manifest_path: PathBuf, + + /// SubnetId of the subnet the artifacts belong to. + #[clap(long, value_parser=ic_recovery::util::subnet_id_from_str)] + subnet_id: SubnetId, +} + +#[allow(clippy::large_enum_variant)] +#[derive(Parser)] +enum Subcommand { + /// Perform Subnet Merging + Merge(MergeArgs), + + /// Validate artifacts produced during subnet merging + Validate(ValidateArgs), +} + +#[derive(Parser)] +#[clap(version = "1.0")] +struct SubnetMergingToolArgs { + #[clap(subcommand)] + subcommand: Subcommand, +} + +fn subnet_merging( + logger: Logger, + recovery_args: RecoveryArgs, + subnet_merging_args: SubnetMergingArgs, + mut neuron_args: Option, +) { + cli::print_step(&logger, "Subnet Merging"); + + info!( + logger, + "Merging subnet with id {} into subnet with id {}", + subnet_merging_args.source_subnet_id, + subnet_merging_args.destination_subnet_id, + ); + warn!( + logger, + "Don't forget to announce at the forum the upcoming series of proposals to merge the \ + subnet" + ); + warn!( + logger, + "See the template at: {}", FORUM_ANNOUNCEMENT_TEMPLATE_URL + ); + + if !recovery_args.skip_prompts { + cli::wait_for_confirmation(&logger); + } + + if neuron_args.is_none() && !recovery_args.test_mode { + neuron_args = Some(cli::read_neuron_args(&logger)); + } + + let subnet_merging = SubnetMerging::new( + logger.clone(), + recovery_args.clone(), + neuron_args, + subnet_merging_args, + ); + + cli::execute_steps(&logger, recovery_args.skip_prompts, subnet_merging); +} + +fn do_merge(args: MergeArgs, logger: Logger) -> anyhow::Result<()> { + let recovery_args = RecoveryArgs { + dir: args.dir, + nns_url: args.nns_url, + replica_version: args.replica_version, + admin_key_file: args.admin_key_file, + test_mode: args.test, + skip_prompts: args.skip_prompts, + }; + + let subnet_merging_state = + cli::read_and_maybe_update_state(&logger, recovery_args, Some(args.subnet_merging_args)); + + subnet_merging( + logger, + subnet_merging_state.recovery_args, + subnet_merging_state.subcommand_args, + subnet_merging_state.neuron_args, + ); + + Ok(()) +} + +fn do_validate(args: ValidateArgs, logger: Logger) -> anyhow::Result<()> { + validate_artifacts( + args.state_tree_path, + args.nns_public_key_path.as_deref(), + args.cup_path, + args.state_manifest_path, + args.subnet_id, + &logger, + ) + .context("Failed to validate the artifacts") +} + +fn main() -> anyhow::Result<()> { + let args = SubnetMergingToolArgs::parse(); + + let logger = util::make_logger(); + + match args.subcommand { + Subcommand::Merge(merge_args) => do_merge(merge_args, logger), + Subcommand::Validate(validate_args) => do_validate(validate_args, logger), + } +} diff --git a/rs/recovery/subnet_merging/src/metrics_helper.rs b/rs/recovery/subnet_merging/src/metrics_helper.rs new file mode 100644 index 000000000000..5c9aa1dc882d --- /dev/null +++ b/rs/recovery/subnet_merging/src/metrics_helper.rs @@ -0,0 +1,219 @@ +//! Scraping of replica metrics, as the `Subnet merging` dashboard reads them. +//! +//! The dashboard evaluates its conditions on Prometheus queries over the +//! metrics of all replicas of a subnet; this module provides the same data to +//! the tool, by scraping the metrics endpoints of the nodes directly. + +use futures::future::join_all; +use ic_recovery::util::block_on; +use slog::{Logger, warn}; + +use std::{collections::BTreeMap, net::IpAddr, time::Duration}; + +/// Timeout of a single metrics request, as in `ic_recovery::get_node_metrics`. +const REQUEST_TIMEOUT: Duration = Duration::from_secs(5); + +/// The metrics of a set of nodes, keyed by series (i.e. metric name plus +/// labels), with one value per node reporting the series. +pub type Metrics = BTreeMap>; + +/// Fetches all series of the given metrics from the given nodes. +/// +/// Best effort, exactly like the dashboard: a node that cannot be scraped +/// contributes no value, and a series that no node reports is absent (which +/// every helper below reads as zero). A condition that has to hold on every +/// node of every subnet therefore has to be written so that missing data keeps +/// it unsatisfied -- see `readiness::evaluate_merge_readiness`. +pub fn fetch_metrics(logger: &Logger, node_ips: &[IpAddr], metrics: &[&str]) -> Metrics { + let responses = block_on(join_all( + node_ips.iter().map(|ip| fetch_node_metrics(logger, ip)), + )); + + let mut result = Metrics::new(); + for (ip, body) in node_ips.iter().zip(responses) { + let Some(body) = body else { + warn!(logger, "Failed to scrape the metrics of node {ip}"); + continue; + }; + for (series, value) in parse_metrics(&body, metrics) { + result.entry(series).or_default().push(value); + } + } + result +} + +async fn fetch_node_metrics(logger: &Logger, ip: &IpAddr) -> Option { + let response = + tokio::time::timeout(REQUEST_TIMEOUT, reqwest::get(format!("http://[{ip}]:9090"))).await; + match response { + Ok(Ok(response)) => match response.text().await { + Ok(body) => Some(body), + Err(err) => { + warn!(logger, "Failed to decode the metrics of node {ip}: {err}"); + None + } + }, + Ok(Err(err)) => { + warn!(logger, "Failed to request the metrics of node {ip}: {err}"); + None + } + Err(_) => { + warn!(logger, "Timed out requesting the metrics of node {ip}"); + None + } + } +} + +/// Picks the series of the requested metrics out of a Prometheus text exposition. +fn parse_metrics(body: &str, metrics: &[&str]) -> Vec<(String, f64)> { + body.lines() + .filter(|line| !line.starts_with('#')) + .filter_map(|line| { + let (series, value) = line.rsplit_once(' ')?; + let series = series.trim(); + if !metrics.iter().any(|metric| is_series_of(series, metric)) { + return None; + } + let value = value.trim().parse::().ok()?; + (!value.is_nan()).then(|| (series.to_string(), value)) + }) + .collect() +} + +/// Whether `series` is a series of `metric`, i.e. the metric name followed by +/// its labels (if any). Metric names are prefixes of one another (e.g. +/// `..._messages` and `..._messages_total`), so a plain prefix check would mix +/// up their series. +fn is_series_of(series: &str, metric: &str) -> bool { + match series.strip_prefix(metric) { + Some(labels) => labels.is_empty() || labels.starts_with('{'), + None => false, + } +} + +/// The per-node values of every series of `metric` whose labels (`{...}`, or +/// the empty string for an unlabeled series) match `labels_match`. +pub fn matching_series<'a>( + metrics: &'a Metrics, + metric: &str, + labels_match: impl Fn(&str) -> bool, +) -> Vec<&'a Vec> { + metrics + .iter() + .filter(|(series, _)| match series.strip_prefix(metric) { + Some(labels) if labels.is_empty() || labels.starts_with('{') => labels_match(labels), + _ => false, + }) + .map(|(_, values)| values) + .collect() +} + +/// Prometheus' `quantile(0.5, ...)`: the median of `values`, interpolating +/// between the two middle values if there is an even number of them. `None` iff +/// `values` is empty. +pub fn median(values: &[f64]) -> Option { + if values.is_empty() { + return None; + } + let mut values = values.to_vec(); + values.sort_by(|a, b| a.partial_cmp(b).expect("metric value should not be NaN")); + let middle = (values.len() - 1) as f64 / 2.0; + Some((values[middle.floor() as usize] + values[middle.ceil() as usize]) / 2.0) +} + +/// `sum(quantile by () (0.5, {}))`: the median across +/// the replicas reporting each matching series, summed over those series. +pub fn sum_of_medians(metrics: &Metrics, metric: &str, labels_match: impl Fn(&str) -> bool) -> f64 { + matching_series(metrics, metric, labels_match) + .into_iter() + .filter_map(|values| median(values)) + .sum() +} + +/// `quantile(0.5, {})`: the median across all replicas +/// reporting any matching series. `None` if there is no such series. +pub fn median_across_replicas( + metrics: &Metrics, + metric: &str, + labels_match: impl Fn(&str) -> bool, +) -> Option { + let values: Vec = matching_series(metrics, metric, labels_match) + .into_iter() + .flatten() + .copied() + .collect(); + median(&values) +} + +#[cfg(test)] +mod tests { + use super::*; + + const BODY: &str = "\ +# HELP mr_stream_messages Messages in streams. +# TYPE mr_stream_messages gauge +mr_stream_messages{remote=\"subnet_1\"} 3 +mr_stream_messages{remote=\"subnet_2\"} 4 +mr_stream_messages_total 7 +replicated_state_pending_refunds 0 +some_other_metric 12 +"; + + #[test] + fn parse_metrics_test() { + let parsed = parse_metrics( + BODY, + &["mr_stream_messages", "replicated_state_pending_refunds"], + ); + + assert_eq!( + parsed, + vec![ + ("mr_stream_messages{remote=\"subnet_1\"}".to_string(), 3.0), + ("mr_stream_messages{remote=\"subnet_2\"}".to_string(), 4.0), + ("replicated_state_pending_refunds".to_string(), 0.0), + ], + "the series of `mr_stream_messages_total`, which `mr_stream_messages` is a prefix \ + of, must not be picked up", + ); + } + + #[test] + fn median_test() { + assert_eq!(median(&[]), None); + assert_eq!(median(&[3.0]), Some(3.0)); + assert_eq!(median(&[3.0, 1.0]), Some(2.0)); + assert_eq!(median(&[3.0, 1.0, 2.0]), Some(2.0)); + assert_eq!(median(&[4.0, 1.0, 3.0, 2.0]), Some(2.5)); + } + + #[test] + fn medians_across_series_test() { + let metrics = Metrics::from([ + ( + "mr_stream_messages{remote=\"a\"}".to_string(), + vec![1.0, 3.0], + ), + ("mr_stream_messages{remote=\"b\"}".to_string(), vec![5.0]), + ("mr_stream_messages_total".to_string(), vec![100.0]), + ]); + + assert_eq!( + sum_of_medians(&metrics, "mr_stream_messages", |_| true), + 7.0 + ); + assert_eq!( + sum_of_medians(&metrics, "mr_stream_messages", |labels| labels + .contains("remote=\"b\"")), + 5.0 + ); + assert_eq!( + median_across_replicas(&metrics, "mr_stream_messages", |_| true), + Some(3.0) + ); + assert_eq!( + median_across_replicas(&metrics, "mr_registry_version", |_| true), + None + ); + } +} diff --git a/rs/recovery/subnet_merging/src/readiness.rs b/rs/recovery/subnet_merging/src/readiness.rs new file mode 100644 index 000000000000..86556dad52b7 --- /dev/null +++ b/rs/recovery/subnet_merging/src/readiness.rs @@ -0,0 +1,362 @@ +//! The "merge readiness" condition of the `Subnet merging` dashboard, evaluated +//! from the metrics of the replicas rather than from a Grafana panel. +//! +//! A subnet that is cooling down is ready to be merged once it has come to a +//! complete rest: every subnet has observed that it is cooling down, nothing is +//! in flight to or from it, and it holds no state that only it could act upon. +//! The terms below spell that out, in the order the dashboard states them. + +use crate::metrics_helper::{self, Metrics}; + +use ic_base_types::SubnetId; +use ic_recovery::{ + error::{RecoveryError, RecoveryResult}, + get_member_node_ids_and_ips, + registry_helper::RegistryHelper, +}; +use ic_registry_client_helpers::subnet::SubnetListRegistry; +use slog::{Logger, info, warn}; + +use std::{ + collections::BTreeMap, + net::IpAddr, + thread::sleep, + time::{Duration, Instant}, +}; + +/// The nodes of every subnet the conditions below range over. +pub type SubnetNodeIps = BTreeMap>; + +/// The registry version every subnet has to have observed. +const METRIC_REGISTRY_VERSION: &str = "mr_registry_version"; +/// The messages held in the streams of a subnet, by remote subnet (the subnet +/// itself included, i.e. its loopback stream). +const METRIC_STREAM_MESSAGES: &str = "mr_stream_messages"; +/// The entries of the ingress history, by status. +const METRIC_INGRESS_HISTORY_BY_STATE: &str = "replicated_state_ingress_history_length_by_state"; +/// The messages held in the input and output queues of the subnet itself, i.e. +/// the management canister's. +const METRIC_SUBNET_INPUT_QUEUE_MESSAGES: &str = "execution_subnet_input_queue_messages"; +const METRIC_SUBNET_OUTPUT_QUEUE_MESSAGES: &str = "execution_subnet_output_queue_messages"; +/// The call contexts of the subnet call context manager, e.g. the `install_code` +/// calls that are still running. +const METRIC_SUBNET_CALL_CONTEXTS: &str = "replicated_state_subnet_call_contexts"; +/// The number of pending anonymous refunds, i.e. the size of the refund pool. +const METRIC_PENDING_REFUNDS: &str = "replicated_state_pending_refunds"; +/// The total value of those refunds, which is logged next to the term above +/// when the refund pool is not empty, but is not itself a condition. +const METRIC_PENDING_REFUNDS_CYCLES: &str = "replicated_state_pending_refunds_cycles"; + +/// A term of the readiness condition: what has to hold, and whether it does. +pub struct Term { + pub description: String, + pub satisfied: bool, +} + +/// Evaluates the terms of the "merge readiness" condition of the `Subnet +/// merging` dashboard for the subnet that is cooling down and the registry +/// version `V` at which it was labeled as such. +/// +/// As in the dashboard, every term is evaluated on the median across the +/// replicas reporting the respective series, and missing data reads as zero +/// (the dashboard's `or vector(0)` fallback). The first term is the one that +/// keeps an unreachable subnet from reading as ready: a subnet whose metrics +/// cannot be scraped reports no registry version, i.e. zero, which is below +/// `V`. +pub fn evaluate_merge_readiness( + subnets: &SubnetNodeIps, + source_subnet_id: SubnetId, + registry_version: u64, + logger: &Logger, +) -> Vec { + let own_metrics = subnet_metrics( + subnets, + source_subnet_id, + &[ + METRIC_REGISTRY_VERSION, + METRIC_STREAM_MESSAGES, + METRIC_INGRESS_HISTORY_BY_STATE, + METRIC_SUBNET_INPUT_QUEUE_MESSAGES, + METRIC_SUBNET_OUTPUT_QUEUE_MESSAGES, + METRIC_SUBNET_CALL_CONTEXTS, + METRIC_PENDING_REFUNDS, + METRIC_PENDING_REFUNDS_CYCLES, + ], + logger, + ); + + // Terms 1 and 2 range over all subnets: the registry version of every + // subnet and the streams of all remote subnets towards this one. + let remote_label = format!("remote=\"{source_subnet_id}\""); + let mut min_registry_version = None; + let mut incoming_stream_messages = 0.0; + for &subnet_id in subnets.keys() { + let metrics = if subnet_id == source_subnet_id { + own_metrics.clone() + } else { + subnet_metrics( + subnets, + subnet_id, + &[METRIC_REGISTRY_VERSION, METRIC_STREAM_MESSAGES], + logger, + ) + }; + let version = + metrics_helper::median_across_replicas(&metrics, METRIC_REGISTRY_VERSION, |_| true) + .unwrap_or(0.0); + min_registry_version = Some(min_registry_version.map_or(version, |v: f64| v.min(version))); + if subnet_id != source_subnet_id { + incoming_stream_messages += + metrics_helper::sum_of_medians(&metrics, METRIC_STREAM_MESSAGES, |labels| { + labels.contains(&remote_label) + }); + } + } + let min_registry_version = min_registry_version.unwrap_or(0.0); + + let outgoing_stream_messages = + metrics_helper::sum_of_medians(&own_metrics, METRIC_STREAM_MESSAGES, |_| true); + let ingress_history_messages = + metrics_helper::sum_of_medians(&own_metrics, METRIC_INGRESS_HISTORY_BY_STATE, |labels| { + !labels.contains("state=\"processing\"") + }); + let subnet_input_queue_messages = + metrics_helper::sum_of_medians(&own_metrics, METRIC_SUBNET_INPUT_QUEUE_MESSAGES, |_| true); + let subnet_output_queue_messages = metrics_helper::median_across_replicas( + &own_metrics, + METRIC_SUBNET_OUTPUT_QUEUE_MESSAGES, + |_| true, + ) + .unwrap_or(0.0); + let subnet_call_contexts = + metrics_helper::sum_of_medians(&own_metrics, METRIC_SUBNET_CALL_CONTEXTS, |_| true); + let pending_refunds = + metrics_helper::median_across_replicas(&own_metrics, METRIC_PENDING_REFUNDS, |_| true) + .unwrap_or(0.0); + let pending_refunds_cycles = + metrics_helper::median_across_replicas(&own_metrics, METRIC_PENDING_REFUNDS_CYCLES, |_| { + true + }) + .unwrap_or(0.0); + + let term = |description: String, satisfied: bool| Term { + description, + satisfied, + }; + vec![ + term( + format!( + "every subnet has reached registry version {registry_version} (the lowest one is \ + at {min_registry_version})" + ), + min_registry_version >= registry_version as f64, + ), + term( + format!( + "no remote subnet holds a message in its stream to subnet {source_subnet_id} \ + ({incoming_stream_messages} messages)" + ), + incoming_stream_messages == 0.0, + ), + term( + format!( + "subnet {source_subnet_id} holds no message in any of its streams, loopback \ + included ({outgoing_stream_messages} messages)" + ), + outgoing_stream_messages == 0.0, + ), + term( + format!( + "the ingress history holds nothing but `processing` entries \ + ({ingress_history_messages} other entries)" + ), + ingress_history_messages == 0.0, + ), + term( + format!("the subnet input queues are empty ({subnet_input_queue_messages} messages)"), + subnet_input_queue_messages == 0.0, + ), + term( + format!("the subnet output queues are empty ({subnet_output_queue_messages} messages)"), + subnet_output_queue_messages == 0.0, + ), + term( + format!( + "the subnet call context manager holds no call context ({subnet_call_contexts} \ + call contexts)" + ), + subnet_call_contexts == 0.0, + ), + term( + // A cooling down subnet routes no refunds either (see `route_refunds` + // in `rs/messaging/src/routing/stream_builder.rs`), so a refund that + // is in the pool stays pending until the subnet is merged, and is + // then lost: the merged state takes the refunds of the destination + // subnet, not those of the subnet that is merged away. + format!( + "the refund pool holds no pending anonymous refund ({pending_refunds} refunds, \ + worth {pending_refunds_cycles} cycles)" + ), + pending_refunds == 0.0, + ), + ] +} + +/// Waits until every term of the readiness condition holds, logging the ones +/// that do not after every round. +pub fn await_merge_readiness( + registry_helper: &RegistryHelper, + source_subnet_id: SubnetId, + registry_version: u64, + timeout: Duration, + poll_interval: Duration, + logger: &Logger, +) -> RecoveryResult<()> { + let deadline = Instant::now() + timeout; + loop { + let terms = evaluate_merge_readiness( + &subnet_node_ips(registry_helper)?, + source_subnet_id, + registry_version, + logger, + ); + let unsatisfied: Vec<&Term> = terms.iter().filter(|term| !term.satisfied).collect(); + + if unsatisfied.is_empty() { + info!( + logger, + "Subnet {source_subnet_id} is ready to be merged; every term holds:" + ); + for term in &terms { + info!(logger, " [x] {}", term.description); + } + return Ok(()); + } + + info!( + logger, + "Subnet {source_subnet_id} is not ready to be merged yet, {} of {} terms do not hold:", + unsatisfied.len(), + terms.len(), + ); + for term in &terms { + let mark = if term.satisfied { "x" } else { " " }; + info!(logger, " [{mark}] {}", term.description); + } + + if Instant::now() >= deadline { + return Err(RecoveryError::UnexpectedError(format!( + "Subnet {source_subnet_id} did not become ready to be merged within {timeout:?}; \ + the terms that do not hold: {}", + unsatisfied + .iter() + .map(|term| term.description.clone()) + .collect::>() + .join("; "), + ))); + } + sleep(poll_interval); + } +} + +/// Waits until every subnet other than `skipped` has reached `registry_version`. +/// +/// `skipped` is the subnet that was merged away: its replicas are stopped by +/// the time this runs, so it reports no metrics anymore, and it is about to be +/// deleted. What matters is that every *other* subnet already routes its +/// canisters to the destination subnet. +pub fn await_registry_version_on_all_subnets( + registry_helper: &RegistryHelper, + skipped: SubnetId, + registry_version: u64, + timeout: Duration, + poll_interval: Duration, + logger: &Logger, +) -> RecoveryResult<()> { + let deadline = Instant::now() + timeout; + loop { + let mut behind = Vec::new(); + let subnets = subnet_node_ips(registry_helper)?; + for &subnet_id in subnets.keys() { + if subnet_id == skipped { + continue; + } + let metrics = subnet_metrics(&subnets, subnet_id, &[METRIC_REGISTRY_VERSION], logger); + let version = + metrics_helper::median_across_replicas(&metrics, METRIC_REGISTRY_VERSION, |_| true) + .unwrap_or(0.0); + if version < registry_version as f64 { + behind.push(format!("{subnet_id} is at registry version {version}")); + } + } + + if behind.is_empty() { + info!( + logger, + "All subnets other than {skipped} have reached registry version {registry_version}" + ); + return Ok(()); + } + + info!( + logger, + "Waiting until all subnets reached registry version {registry_version}: {}", + behind.join("; "), + ); + + if Instant::now() >= deadline { + return Err(RecoveryError::UnexpectedError(format!( + "Not all subnets reached registry version {registry_version} within {timeout:?}: \ + {}", + behind.join("; "), + ))); + } + sleep(poll_interval); + } +} + +/// The nodes of every subnet in the registry. +pub fn subnet_node_ips(registry_helper: &RegistryHelper) -> RecoveryResult { + let registry_version = registry_helper.latest_registry_version()?; + let subnet_ids = registry_helper + .registry_client() + .get_subnet_ids(registry_version) + .map_err(|err| { + RecoveryError::RegistryError(format!( + "Failed to get the subnet ids at registry version {registry_version}: {err}" + )) + })? + .ok_or_else(|| { + RecoveryError::RegistryError(format!( + "No subnet ids at registry version {registry_version}" + )) + })?; + + subnet_ids + .into_iter() + .map(|subnet_id| { + let node_ips = get_member_node_ids_and_ips(registry_helper, subnet_id)? + .into_values() + .collect(); + Ok((subnet_id, node_ips)) + }) + .collect() +} + +/// Fetches the given metrics from all nodes of `subnet_id`. +fn subnet_metrics( + subnets: &SubnetNodeIps, + subnet_id: SubnetId, + metrics: &[&str], + logger: &Logger, +) -> Metrics { + let node_ips = match subnets.get(&subnet_id) { + Some(node_ips) if !node_ips.is_empty() => node_ips.as_slice(), + _ => { + warn!(logger, "Subnet {subnet_id} has no node to scrape"); + &[] + } + }; + + metrics_helper::fetch_metrics(logger, node_ips, metrics) +} diff --git a/rs/recovery/subnet_merging/src/steps.rs b/rs/recovery/subnet_merging/src/steps.rs new file mode 100644 index 000000000000..b7d06764d916 --- /dev/null +++ b/rs/recovery/subnet_merging/src/steps.rs @@ -0,0 +1,673 @@ +use crate::{ + layout::{CUP_FILE_NAME, Layout}, + readiness, + target_subnet::TargetSubnet, + utils::{MergedStateParams, first_registry_version_where}, +}; + +use ic_base_types::SubnetId; +use ic_recovery::{ + IC_DATA_PATH, Recovery, STATES_METADATA, + error::{RecoveryError, RecoveryResult}, + get_node_metrics, + registry_helper::RegistryHelper, + ssh_helper::SshHelper, + steps::Step, + util::block_on, +}; +use ic_registry_client_helpers::routing_table::RoutingTableRegistry; +use ic_registry_routing_table::RoutingTable; +use ic_subnet_tools::{ + agent_helper::AgentHelper, + state_tool_helper, + utils::{get_cup, get_state_hash}, + validation::validate_artifacts, +}; +use ic_types::{Height, consensus::CatchUpPackage, consensus::HasHeight}; +use slog::{Logger, info, warn}; +use url::Url; + +use std::{ + net::IpAddr, + path::PathBuf, + thread::sleep, + time::{Duration, Instant}, +}; + +/// Retries `check` until it reports success, `timeout` elapses, or it fails +/// with an error that is not worth retrying. +/// +/// What the steps below use it for is the registry catching up with a proposal +/// that `ic-admin` just reported as executed: the local store is polled on +/// every read, but the mutation may not have made it into the registry canister +/// the poll reads from yet. +fn wait_for( + what: &str, + timeout: Duration, + poll_interval: Duration, + logger: &Logger, + check: impl Fn() -> RecoveryResult, +) -> RecoveryResult<()> { + let deadline = Instant::now() + timeout; + loop { + match check() { + Ok(true) => return Ok(()), + Ok(false) => info!(logger, "Waiting until {what}"), + Err(err) => warn!(logger, "Waiting until {what}, last attempt failed: {err}"), + } + + if Instant::now() >= deadline { + return Err(RecoveryError::UnexpectedError(format!( + "Gave up waiting until {what} after {timeout:?}" + ))); + } + sleep(poll_interval); + } +} + +/// Reads the record of the subnet that is being merged away, checks that it is +/// labeled "cooling down" and determines (and persists) the registry version +/// `V` at which it was labeled so, which is what the merge readiness condition +/// is evaluated against. +pub(crate) struct CheckCoolingDownStep { + pub(crate) source_subnet_id: SubnetId, + pub(crate) registry_helper: RegistryHelper, + pub(crate) layout: Layout, + pub(crate) timeout: Duration, + pub(crate) poll_interval: Duration, + pub(crate) logger: Logger, +} + +impl Step for CheckCoolingDownStep { + fn descr(&self) -> String { + format!( + "Read the registry to check that subnet {} is labeled \"cooling down\", and determine \ + the registry version it was labeled so at.", + self.source_subnet_id, + ) + } + + fn exec(&self) -> RecoveryResult<()> { + let source_subnet_id = self.source_subnet_id; + wait_for( + &format!("subnet {source_subnet_id} is labeled \"cooling down\""), + self.timeout, + self.poll_interval, + &self.logger, + || { + let (registry_version, subnet_record) = + self.registry_helper.get_subnet_record(source_subnet_id)?; + let Some(subnet_record) = subnet_record else { + return Err(RecoveryError::RegistryError(format!( + "No record of subnet {source_subnet_id} at registry version \ + {registry_version}" + ))); + }; + Ok(subnet_record.cooling_down) + }, + )?; + + let registry_helper = self.registry_helper.clone(); + let cooling_down_version = + first_registry_version_where(&self.registry_helper, |version| { + Ok(registry_helper + .get_subnet_record_at_version(source_subnet_id, version)? + .is_some_and(|record| record.cooling_down)) + })?; + + info!( + self.logger, + "Subnet {source_subnet_id} has been labeled \"cooling down\" since registry version \ + {cooling_down_version}", + ); + + self.layout + .write_cooling_down_registry_version(cooling_down_version.get()) + } +} + +/// Waits until the merge readiness condition of the `Subnet merging` dashboard +/// holds for the subnet that is cooling down. +pub(crate) struct CheckMergeReadinessStep { + pub(crate) source_subnet_id: SubnetId, + pub(crate) registry_helper: RegistryHelper, + pub(crate) layout: Layout, + pub(crate) timeout: Duration, + pub(crate) poll_interval: Duration, + pub(crate) logger: Logger, +} + +impl Step for CheckMergeReadinessStep { + fn descr(&self) -> String { + format!( + "Wait until subnet {} is ready to be merged, i.e. until it has come to a complete \ + rest: every subnet has observed that it is cooling down, no message is in flight to \ + or from it, its ingress history holds nothing but `processing` entries, its subnet \ + queues and call context manager are empty and its refund pool holds no pending \ + anonymous refund.", + self.source_subnet_id, + ) + } + + fn exec(&self) -> RecoveryResult<()> { + let registry_version = self.layout.read_cooling_down_registry_version()?; + + readiness::await_merge_readiness( + &self.registry_helper, + self.source_subnet_id, + registry_version, + self.timeout, + self.poll_interval, + &self.logger, + ) + } +} + +/// Waits until the node the state is downloaded from holds the CUP its subnet +/// halts at, i.e. the first one whose summary is created at a registry version +/// carrying the `halt_at_cup_height` flag. +/// +/// Waiting for the CUP rather than for the subnet to report that it is halted: +/// the CUP is what names the state the subnet came to rest in, and it exists +/// only once that state has been certified and its hash agreed upon. A subnet +/// that has just stopped delivering batches, on the other hand, may not have +/// finished writing and hashing the checkpoint it stopped at, and downloading +/// its latest checkpoint then yields the previous one, a whole DKG interval +/// before the state the merge is supposed to be assembled from. +/// +/// The CUP the node serves over HTTP is not enough: the orchestrator writes the +/// CUP to disk asynchronously, and it is the on-disk copy that is downloaded +/// with the state and validated afterwards. So this waits for both, and for +/// them to agree. +pub(crate) struct WaitForHaltingCupStep { + pub(crate) subnet_id: SubnetId, + pub(crate) node_ip: IpAddr, + pub(crate) registry_helper: RegistryHelper, + pub(crate) layout: Layout, + pub(crate) ssh_helper: SshHelper, + pub(crate) timeout: Duration, + pub(crate) poll_interval: Duration, + pub(crate) logger: Logger, +} + +impl Step for WaitForHaltingCupStep { + fn descr(&self) -> String { + format!( + "Wait until node {} holds the CUP subnet {} halts at, both at its public endpoint and \ + on disk, and until it has certified the state that CUP names.", + self.node_ip, self.subnet_id, + ) + } + + fn exec(&self) -> RecoveryResult<()> { + let deadline = Instant::now() + self.timeout; + loop { + match self.check() { + Ok(Some(height)) => { + info!( + self.logger, + "Subnet {} halted at height {height}", self.subnet_id + ); + return Ok(()); + } + Ok(None) => {} + // A node that is not answering yet, or a CUP that cannot be + // pulled off it yet, is the normal state of affairs while the + // subnet is still running towards its halting CUP. + Err(err) => warn!( + self.logger, + "Subnet {} has not halted yet: {err}", self.subnet_id + ), + } + + if Instant::now() >= deadline { + return Err(RecoveryError::UnexpectedError(format!( + "Subnet {} did not reach the CUP it halts at within {:?}", + self.subnet_id, self.timeout, + ))); + } + sleep(self.poll_interval); + } + } +} + +impl WaitForHaltingCupStep { + /// Returns the height of the halting CUP if the node has reached it, `None` + /// if it has not yet, and an error if the node could not be asked. + fn check(&self) -> RecoveryResult> { + let cup = self.served_cup()?; + let cup_height = cup.height(); + let cup_registry_version = cup + .content + .block + .get_value() + .payload + .as_ref() + .as_summary() + .dkg + .registry_version; + + // The `halt_at_cup_height` flag is read at the registry version of the + // summary block active at a height, and that version only changes at a + // summary, so batch delivery stops exactly when the summary carrying the + // flag becomes active. + let halting = self + .registry_helper + .get_subnet_record_at_version(self.subnet_id, cup_registry_version)? + .is_some_and(|record| record.halt_at_cup_height); + if !halting { + info!( + self.logger, + "Subnet {} is at the CUP at height {cup_height}, whose registry version \ + {cup_registry_version} does not carry the `halt_at_cup_height` flag yet", + self.subnet_id, + ); + return Ok(None); + } + + // The node has to have caught up with the CUP itself: it is its state + // that is downloaded, and a node can hold a CUP that the rest of the + // subnet assembled before it got there. + let metrics = block_on(get_node_metrics(&self.logger, &self.node_ip)).ok_or_else(|| { + RecoveryError::UnexpectedError(format!( + "Failed to get the metrics of node {}", + self.node_ip + )) + })?; + if metrics.certification_height > cup_height { + return Err(RecoveryError::ValidationFailed(format!( + "Subnet {} certified height {}, past the CUP at height {cup_height} it should \ + have halted at", + self.subnet_id, metrics.certification_height, + ))); + } + if metrics.certification_height < cup_height { + info!( + self.logger, + "Subnet {} holds the CUP at height {cup_height} but node {} has only certified up \ + to height {}", + self.subnet_id, + self.node_ip, + metrics.certification_height, + ); + return Ok(None); + } + + // The CUP the orchestrator has written to disk is the one that is + // downloaded with the state and validated afterwards, so it has to be + // the halting CUP, too. + let on_disk_cup = self.on_disk_cup()?; + if on_disk_cup.height() != cup_height + || on_disk_cup.content.state_hash != cup.content.state_hash + { + info!( + self.logger, + "Node {} still holds the CUP at height {} on disk, not the one at height \ + {cup_height} it serves", + self.node_ip, + on_disk_cup.height(), + ); + return Ok(None); + } + + Ok(Some(cup_height)) + } + + /// The CUP the node serves at its public endpoint. + fn served_cup(&self) -> RecoveryResult { + let url = Url::parse(&format!("http://[{}]:8080/", self.node_ip)).map_err(|err| { + RecoveryError::UnexpectedError(format!( + "Could not parse the URL of node {}: {err}", + self.node_ip + )) + })?; + + let cup_proto = block_on(ic_cup_explorer::get_cup(&url)) + .map_err(|err| { + RecoveryError::UnexpectedError(format!( + "Failed to get the CUP of node {}: {err}", + self.node_ip + )) + })? + .ok_or_else(|| { + RecoveryError::UnexpectedError(format!("Node {} serves no CUP", self.node_ip)) + })?; + + CatchUpPackage::try_from(&cup_proto).map_err(|err| { + RecoveryError::UnexpectedError(format!( + "Failed to deserialize the CUP of node {}: {err}", + self.node_ip + )) + }) + } + + /// The CUP the orchestrator of the node has written to disk, pulled off the + /// node and kept next to the other artifacts of the merge. + fn on_disk_cup(&self) -> RecoveryResult { + let remote_cup = self.ssh_helper.remote_path( + PathBuf::from(IC_DATA_PATH) + .join(ic_recovery::CUPS_DIR) + .join(CUP_FILE_NAME), + ); + let local_cup = self.layout.halting_cup_file(self.subnet_id); + + self.ssh_helper.rsync(remote_cup, &local_cup)?; + + get_cup(&local_cup) + } +} + +/// Validates the CUP a subnet halted at and the state that was downloaded with +/// it: the subnet's public key is taken from the NNS signed state tree, the CUP +/// signature is verified against it, and the manifest recomputed from the +/// downloaded state has to match the state hash the CUP names. +pub(crate) struct ValidateCupStep { + pub(crate) subnet_id: SubnetId, + pub(crate) target_subnet: TargetSubnet, + pub(crate) nns_url: Url, + pub(crate) layout: Layout, + pub(crate) logger: Logger, +} + +impl Step for ValidateCupStep { + fn descr(&self) -> String { + format!( + "Validate the CUP and the state downloaded from the {} subnet {}, and preserve the \ + subnet's public key and the state tree (with only the relevant paths) so that they \ + can be verified independently.", + self.target_subnet, self.subnet_id, + ) + } + + fn exec(&self) -> RecoveryResult<()> { + // 1. Get the subnet's public key using `ic-agent` and persist it on disk. + info!(self.logger, "Getting the NNS signed State Tree"); + let agent_helper = AgentHelper::new( + &self.nns_url, + Some(self.layout.nns_public_key_file()), + self.logger.clone(), + )?; + + let pruned_state_tree = agent_helper.read_subnet_data(self.subnet_id)?; + pruned_state_tree.save_to_file(&self.layout.pruned_state_tree_file(self.subnet_id))?; + pruned_state_tree + .save_public_key_to_file(&self.layout.subnet_public_key_file(self.subnet_id))?; + + // 2. Compute the manifest of the downloaded state. + info!(self.logger, "Computing the state manifest"); + let checkpoint_dir = self.layout.latest_checkpoint_dir(self.target_subnet)?; + let manifest_path = self.layout.actual_manifest_file(self.subnet_id); + + state_tool_helper::compute_manifest(&checkpoint_dir, &manifest_path)?; + + // 3. Validate all the artifacts (state tree, CUP, state manifest). + validate_artifacts( + self.layout.pruned_state_tree_file(self.subnet_id), + Some(self.layout.nns_public_key_file()), + self.layout.downloaded_cup_file(self.target_subnet), + manifest_path, + self.subnet_id, + &self.logger, + ) + } +} + +/// Assembles the merged state from the states the two subnets halted at: the +/// state of the destination subnet, with the canisters and canister snapshots +/// of the source subnet added to it, as a new checkpoint that the destination +/// subnet is then recovered at. +pub(crate) struct MergeStatesStep { + pub(crate) layout: Layout, + pub(crate) time_margin: Duration, + pub(crate) logger: Logger, +} + +impl Step for MergeStatesStep { + fn descr(&self) -> String { + format!( + "Assemble the merged state from the states downloaded to {} and {}, as a new \ + checkpoint in {}, and compute the height, the batch time and the state hash the \ + recovery of the destination subnet needs.", + self.layout.work_dir(TargetSubnet::Source).display(), + self.layout.work_dir(TargetSubnet::Destination).display(), + self.layout.work_dir(TargetSubnet::Merged).display(), + ) + } + + fn exec(&self) -> RecoveryResult<()> { + let source_height = self.layout.latest_checkpoint_height(TargetSubnet::Source)?; + let destination_height = self + .layout + .latest_checkpoint_height(TargetSubnet::Destination)?; + let source_checkpoint = self + .layout + .checkpoint_dir(TargetSubnet::Source, source_height); + let destination_checkpoint = self + .layout + .checkpoint_dir(TargetSubnet::Destination, destination_height); + + // The recovery CUP is created at the next recovery height above the one + // the destination subnet halted at, exactly as in any other recovery. + let merged_height = Recovery::get_recovery_height(destination_height); + let merged_checkpoint = self + .layout + .checkpoint_dir(TargetSubnet::Merged, merged_height); + + // The block time the recovered subnet starts from has to be larger than + // the times of both checkpoints the merged state is assembled from. + let source_time = state_tool_helper::checkpoint_time_nanos(&source_checkpoint)?; + let destination_time = state_tool_helper::checkpoint_time_nanos(&destination_checkpoint)?; + let merged_time = source_time.max(destination_time) + self.time_margin.as_nanos() as u64; + info!( + self.logger, + "The source subnet halted at height {source_height} and time {source_time}, the \ + destination subnet at height {destination_height} and time {destination_time}; the \ + merged state is the checkpoint {merged_height} and starts at time {merged_time}", + ); + + info!(self.logger, "Merging the states"); + state_tool_helper::merge_checkpoints( + &destination_checkpoint, + &source_checkpoint, + &merged_checkpoint, + )?; + + info!(self.logger, "Computing the manifest of the merged state"); + let manifest_path = self.layout.merged_state_manifest_file(); + state_tool_helper::compute_manifest(&merged_checkpoint, manifest_path)?; + + info!(self.logger, "Validating the manifest of the merged state"); + let manifest_hash = state_tool_helper::verify_manifest(manifest_path) + .map_err(|err| RecoveryError::validation_failed("Manifest verification failed", err))?; + let state_hash = get_state_hash(&merged_checkpoint)?; + if manifest_hash != state_hash { + return Err(RecoveryError::ValidationFailed(format!( + "The root hash {manifest_hash} of the manifest of the merged state differs from \ + the hash {state_hash} recomputed from the merged checkpoint", + ))); + } + + // The upload step transfers the states metadata alongside the + // checkpoint, and rsync is given every path it transfers as an explicit + // source, so a missing one fails the whole transfer: take the + // destination subnet's along. + // + // It is a manifest cache, which the state manager recomputes for the + // checkpoints it finds whenever it is missing or does not describe them, + // and the heights it names here are the ones the destination subnet held + // before the merge, none of which the merged state directory has. That is + // the same mismatch a plain subnet recovery uploads, where the metadata + // comes from the state that was downloaded and the checkpoint from the + // replay that followed. + std::fs::copy( + self.layout + .ic_state_dir(TargetSubnet::Destination) + .join(STATES_METADATA), + self.layout + .ic_state_dir(TargetSubnet::Merged) + .join(STATES_METADATA), + ) + .map_err(|err| { + RecoveryError::UnexpectedError(format!("Failed to copy the states metadata: {err}")) + })?; + + let params = MergedStateParams { + height: merged_height.get(), + time_nanos: merged_time, + state_hash, + }; + info!(self.logger, "The merged state: {params:?}"); + + params.write(self.layout.merged_state_params_file()) + } +} + +/// Reads the routing table, checks that the canister ID ranges of the subnet +/// that was merged away are hosted by the destination subnet now, and +/// determines (and persists) the registry version the merge was applied at. +pub(crate) struct CheckRoutingTableStep { + pub(crate) source_subnet_id: SubnetId, + pub(crate) destination_subnet_id: SubnetId, + pub(crate) registry_helper: RegistryHelper, + pub(crate) layout: Layout, + pub(crate) timeout: Duration, + pub(crate) poll_interval: Duration, + pub(crate) logger: Logger, +} + +impl Step for CheckRoutingTableStep { + fn descr(&self) -> String { + format!( + "Read the routing table to check that subnet {} hosts no canister id range anymore \ + and that subnet {} hosts them now.", + self.source_subnet_id, self.destination_subnet_id, + ) + } + + fn exec(&self) -> RecoveryResult<()> { + let source_subnet_id = self.source_subnet_id; + wait_for( + &format!("subnet {source_subnet_id} hosts no canister id range anymore"), + self.timeout, + self.poll_interval, + &self.logger, + || { + let (registry_version, routing_table) = self.registry_helper.get_routing_table()?; + let Some(routing_table) = routing_table else { + return Err(RecoveryError::RegistryError(format!( + "No routing table at registry version {registry_version}" + ))); + }; + Ok(source_subnet_is_empty(&routing_table, source_subnet_id)) + }, + )?; + + let (registry_version, routing_table) = self.registry_helper.get_routing_table()?; + let Some(routing_table) = routing_table else { + return Err(RecoveryError::RegistryError(format!( + "No routing table at registry version {registry_version}" + ))); + }; + info!( + self.logger, + "Canister id ranges of subnet {} at registry version {registry_version}: {:#?}", + self.destination_subnet_id, + routing_table.ranges(self.destination_subnet_id), + ); + + let registry_client = self.registry_helper.registry_client(); + let merge_registry_version = + first_registry_version_where(&self.registry_helper, |version| { + let routing_table = registry_client + .get_routing_table(version) + .map_err(|err| { + RecoveryError::RegistryError(format!( + "Failed to get the routing table at registry version {version}: {err}" + )) + })? + .ok_or_else(|| { + RecoveryError::RegistryError(format!( + "No routing table at registry version {version}" + )) + })?; + Ok(source_subnet_is_empty(&routing_table, source_subnet_id)) + })?; + + info!( + self.logger, + "Subnet {source_subnet_id} has been merged away as of registry version \ + {merge_registry_version}", + ); + + self.layout + .write_merge_registry_version(merge_registry_version.get()) + } +} + +fn source_subnet_is_empty(routing_table: &RoutingTable, source_subnet_id: SubnetId) -> bool { + routing_table.ranges(source_subnet_id).is_empty() +} + +/// Waits until every subnet other than the one that was merged away routes the +/// canisters of the merged subnet to the destination subnet, i.e. until it is +/// safe to delete the merged subnet. +pub(crate) struct CheckRegistryVersionOnAllSubnetsStep { + pub(crate) source_subnet_id: SubnetId, + pub(crate) registry_helper: RegistryHelper, + pub(crate) layout: Layout, + pub(crate) timeout: Duration, + pub(crate) poll_interval: Duration, + pub(crate) logger: Logger, +} + +impl Step for CheckRegistryVersionOnAllSubnetsStep { + fn descr(&self) -> String { + format!( + "Wait until every subnet other than {} has reached the registry version the merge \ + created, i.e. routes the canisters that used to be hosted by it to the destination \ + subnet.", + self.source_subnet_id, + ) + } + + fn exec(&self) -> RecoveryResult<()> { + let registry_version = self.layout.read_merge_registry_version()?; + + readiness::await_registry_version_on_all_subnets( + &self.registry_helper, + self.source_subnet_id, + registry_version, + self.timeout, + self.poll_interval, + &self.logger, + ) + } +} + +/// Waits until the destination subnet came up on the merged state, i.e. until +/// the node the merged state was uploaded to reports the recovery CUP. +pub(crate) struct WaitForRecoveryCupStep { + pub(crate) node_ip: IpAddr, + pub(crate) layout: Layout, + pub(crate) logger: Logger, +} + +impl Step for WaitForRecoveryCupStep { + fn descr(&self) -> String { + format!( + "Wait until node {} reports the recovery CUP holding the merged state.", + self.node_ip, + ) + } + + fn exec(&self) -> RecoveryResult<()> { + let params = MergedStateParams::read(self.layout.merged_state_params_file())?; + + Recovery::wait_for_recovery_cup( + &self.logger, + self.node_ip, + Height::from(params.height), + params.state_hash, + ) + } +} diff --git a/rs/recovery/subnet_merging/src/subnet_merging.rs b/rs/recovery/subnet_merging/src/subnet_merging.rs new file mode 100644 index 000000000000..34f22e7ee366 --- /dev/null +++ b/rs/recovery/subnet_merging/src/subnet_merging.rs @@ -0,0 +1,739 @@ +use crate::{ + admin_helper::{ + get_propose_to_cool_down_subnet_command, get_propose_to_delete_subnet_command, + get_propose_to_merge_subnets_command, + }, + layout::Layout, + steps::{ + CheckCoolingDownStep, CheckMergeReadinessStep, CheckRegistryVersionOnAllSubnetsStep, + CheckRoutingTableStep, MergeStatesStep, ValidateCupStep, WaitForHaltingCupStep, + WaitForRecoveryCupStep, + }, + target_subnet::TargetSubnet, + utils::MergedStateParams, +}; + +use clap::Parser; +use ic_base_types::SubnetId; +use ic_protobuf::registry::subnet::v1::SubnetRecord; +use ic_recovery::{ + CUPS_DIR, NeuronArgs, Recovery, RecoveryArgs, + cli::{consent_given, read_optional}, + error::{RecoveryError, RecoveryResult}, + recovery_iterator::RecoveryIterator, + recovery_state::{HasRecoveryState, RecoveryState}, + registry_helper::RegistryPollingStrategy, + ssh_helper::SshHelper, + steps::{AdminStep, DownloadIcDataStep, Step, UploadStateAndRestartStep}, + util::{CheckpointHeight, DataLocation, ExecutionMode, SshUser}, +}; +use ic_registry_subnet_type::SubnetType; +use ic_subnet_tools::{ + admin_helper::get_halt_subnet_at_cup_height_command, cli::print_url_and_ask_for_confirmation, +}; +use ic_types::Height; +use serde::{Deserialize, Serialize}; +use slog::{Logger, info, warn}; +use strum::{EnumMessage, IntoEnumIterator}; +use strum_macros::{EnumIter, EnumString}; + +use std::{ + iter::Peekable, + net::IpAddr, + path::PathBuf, + time::{Duration, UNIX_EPOCH}, +}; + +const SUBNET_TYPE_ALLOW_LIST: [SubnetType; 2] = + [SubnetType::Application, SubnetType::VerifiedApplication]; + +#[derive( + Copy, + Clone, + PartialEq, + Debug, + Deserialize, + EnumIter, + EnumMessage, + EnumString, + Serialize, + clap::ValueEnum, +)] +pub enum StepType { + CoolDownSourceSubnet, + CheckRegistryForCoolingDownFlag, + CheckMergeReadiness, + HaltSourceSubnetAtCupHeight, + HaltDestinationSubnetAtCupHeight, + WaitForHaltingCupOnSourceSubnet, + WaitForHaltingCupOnDestinationSubnet, + StopSourceReplica, + StopDestinationReplica, + DownloadStateFromSourceSubnet, + DownloadStateFromDestinationSubnet, + ValidateSourceSubnetCup, + ValidateDestinationSubnetCup, + MergeStates, + MergeSubnets, + CheckRegistryForRoutingTableEntry, + ProposeCupForDestinationSubnet, + UploadStateToDestinationSubnet, + WaitForCUPOnDestinationSubnet, + UnhaltDestinationSubnet, + CheckRegistryVersionOnAllSubnets, + DeleteSourceSubnet, + Cleanup, +} + +#[derive(Clone, PartialEq, Debug, Deserialize, Parser, Serialize)] +#[clap(version = "1.0")] +pub struct SubnetMergingArgs { + /// Id of the subnet that is cooling down, whose canisters are merged into + /// the destination subnet and which is deleted afterwards. + #[clap(long, value_parser=ic_recovery::util::subnet_id_from_str)] + pub source_subnet_id: SubnetId, + + /// Id of the subnet that hosts the canisters of the source subnet after the + /// merge, and that is recovered at the merged state. + #[clap(long, value_parser=ic_recovery::util::subnet_id_from_str)] + pub destination_subnet_id: SubnetId, + + /// Public ssh key to be deployed to both subnets for read only access. + #[clap(long)] + pub readonly_pub_key: Option, + + /// The path to a file containing the private key associated with `readonly_pub_key`. + #[clap(long)] + pub readonly_key_file: Option, + + /// If the downloaded states should be backed up locally. + #[clap(long)] + pub keep_downloaded_state: Option, + + /// IP address of the node of the source subnet to download the state from. + #[clap(long)] + pub download_node_source: Option, + + /// IP address of the node of the destination subnet to download the state from. + #[clap(long)] + pub download_node_destination: Option, + + /// IP address of the node of the destination subnet to upload the merged state to. + #[clap(long)] + pub upload_node_destination: Option, + + /// How much later than the checkpoints it is assembled from the merged + /// state starts, in seconds. + #[clap(long, default_value = "60")] + pub time_margin_secs: u64, + + /// How long to wait for the subnet that is cooling down to become ready to + /// be merged, in seconds. + #[clap(long, default_value = "2400")] + pub merge_ready_timeout_secs: u64, + + /// How long to wait for a subnet to reach the CUP it halts at, in seconds. + #[clap(long, default_value = "900")] + pub halt_timeout_secs: u64, + + /// How long to wait for the registry to reflect a proposal that `ic-admin` + /// reported as executed, in seconds. + #[clap(long, default_value = "300")] + pub registry_timeout_secs: u64, + + /// How long to wait between two evaluations of a condition this tool waits + /// for, in seconds. + #[clap(long, default_value = "10")] + pub poll_interval_secs: u64, + + /// If present the tool will start execution for the provided step, skipping the initial ones. + #[clap(long = "resume")] + #[clap(value_enum)] + pub next_step: Option, +} + +pub struct SubnetMerging { + step_iterator: Peekable, + params: SubnetMergingArgs, + recovery_args: RecoveryArgs, + neuron_args: Option, + recovery: Recovery, + layout: Layout, + logger: Logger, +} + +impl SubnetMerging { + pub fn new( + logger: Logger, + recovery_args: RecoveryArgs, + neuron_args: Option, + subnet_merging_args: SubnetMergingArgs, + ) -> Self { + let recovery = Recovery::new( + logger.clone(), + recovery_args.clone(), + neuron_args.clone(), + recovery_args.nns_url.clone(), + RegistryPollingStrategy::WithEveryRead, + ) + .expect("Failed to initialize recovery"); + + Self::check_subnets_preconditions( + &recovery, + subnet_merging_args.source_subnet_id, + subnet_merging_args.destination_subnet_id, + ) + .expect("Subnets should satisfy all the preconditions"); + + let layout = Layout::new(&recovery); + layout + .create_dirs() + .expect("Failed to create the working directories"); + + Self { + step_iterator: StepType::iter().peekable(), + params: subnet_merging_args, + recovery_args, + neuron_args, + layout, + recovery, + logger, + } + } + + /// Checks whether the subnets satisfy the following preconditions: + /// + /// Both subnets: + /// 1) Are `Application` (or `VerifiedApplication`) subnets; + /// 2) Are not Chain key subnets; + /// 3) Are not halted: this tool halts them itself, at their next CUP; + /// 4) Have the same subnet type, as the merged state inherits the + /// destination subnet's. + /// + /// And they are two different subnets. Unlike subnet splitting, the + /// destination subnet is a subnet in operation: it keeps serving its own + /// canisters across the merge, and it may or may not be empty. + fn check_subnets_preconditions( + recovery: &Recovery, + source_subnet_id: SubnetId, + destination_subnet_id: SubnetId, + ) -> RecoveryResult<()> { + if source_subnet_id == destination_subnet_id { + return Err(RecoveryError::ValidationFailed(format!( + "A subnet cannot be merged into itself ({source_subnet_id})" + ))); + } + + let source_subnet_record = + Self::get_and_pre_validate_subnet_record(recovery, source_subnet_id, None)?; + + let _ = Self::get_and_pre_validate_subnet_record( + recovery, + destination_subnet_id, + Some(source_subnet_record), + )?; + + Ok(()) + } + + fn get_and_pre_validate_subnet_record( + recovery: &Recovery, + subnet_id: SubnetId, + other_subnet_record: Option, + ) -> RecoveryResult { + let validation_error = |error_message| { + Err(RecoveryError::ValidationFailed(format!( + "Subnet {subnet_id}: {error_message}" + ))) + }; + + let (_, Some(subnet_record)) = recovery.registry_helper.get_subnet_record(subnet_id)? + else { + return validation_error("Subnet Record should not be empty".to_string()); + }; + + if subnet_record + .chain_key_config + .as_ref() + .is_some_and(|chain_key_config| !chain_key_config.key_configs.is_empty()) + { + return validation_error("Subnet should not be a Chain key subnet".to_string()); + } + + let subnet_type = subnet_record + .subnet_type() + .try_into() + .expect("Unexpected subnet type"); + + if !SUBNET_TYPE_ALLOW_LIST.contains(&subnet_type) { + return validation_error(format!( + "Subnet's type ({subnet_type:?}) is not allowed for subnet merging. Allowlist: {SUBNET_TYPE_ALLOW_LIST:?}", + )); + } + + if subnet_record.is_halted { + return validation_error(String::from( + "Subnet should not be halted: subnet merging halts both subnets itself, at the \ + CUP whose state it merges", + )); + } + + if let Some(other_subnet_record) = other_subnet_record + && subnet_record.subnet_type() != other_subnet_record.subnet_type() + { + return validation_error(format!( + "Both subnets should have the same subnet type. \ + Expected subnet type = {:?}, actual subnet type = {:?}", + other_subnet_record.subnet_type(), + subnet_record.subnet_type(), + )); + } + + Ok(subnet_record) + } + + fn wait_for_halting_cup_step( + &self, + target_subnet: TargetSubnet, + ) -> RecoveryResult> { + let Some(node_ip) = self.download_node(target_subnet) else { + return Err(RecoveryError::StepSkipped); + }; + + Ok(WaitForHaltingCupStep { + subnet_id: self.subnet_id(target_subnet), + node_ip, + registry_helper: self.recovery.registry_helper.clone(), + layout: self.layout.clone(), + ssh_helper: self.ssh_helper(node_ip), + timeout: Duration::from_secs(self.params.halt_timeout_secs), + poll_interval: Duration::from_secs(self.params.poll_interval_secs), + logger: self.recovery.logger.clone(), + }) + } + + fn stop_replica_step(&self, target_subnet: TargetSubnet) -> RecoveryResult> { + // The state manager of a running replica owns its state directory, even + // while consensus is halted, so the replica has to be stopped before the + // state is downloaded. The destination subnet's replica stays stopped + // until the merged state is uploaded to it; the source subnet's stays + // stopped for good, as that subnet is deleted at the end of the merge. + match self.download_node(target_subnet) { + Some(node_ip) => Ok(self.recovery.get_stop_replica_step(node_ip)), + None => Err(RecoveryError::StepSkipped), + } + } + + /// Downloads the state a subnet halted at into that subnet's working + /// directory. + /// + /// `DownloadIcDataStep` rather than `Recovery::get_download_data_step`: a + /// merge downloads two states, and the latter always downloads into the + /// single working directory a `Recovery` has. + fn download_state_step( + &self, + target_subnet: TargetSubnet, + ) -> RecoveryResult> { + let Some(node_ip) = self.download_node(target_subnet) else { + return Err(RecoveryError::StepSkipped); + }; + + let mut ssh_helper = self.ssh_helper(node_ip); + if ssh_helper.wait_for_access().is_err() { + ssh_helper.ssh_user = SshUser::Admin; + ssh_helper.key_file = self.recovery.admin_key_file.clone(); + if !ssh_helper.can_connect() { + return Err(RecoveryError::UnexpectedError(format!( + "SSH access to node {node_ip} denied" + ))); + } + } + info!( + self.recovery.logger, + "Continuing with account: {}", ssh_helper.ssh_user + ); + + let mut includes = Recovery::get_ic_state_includes( + &self.recovery.logger, + ExecutionMode::Remote(&ssh_helper), + CheckpointHeight::Latest, + )?; + // The CUP the subnet halted at, which the validation step checks the + // downloaded state against. + includes.push(PathBuf::from(CUPS_DIR)); + + Ok(DownloadIcDataStep { + logger: self.recovery.logger.clone(), + ssh_helper, + backup_dir: self.layout.original_data_dir(target_subnet), + working_dir: self.layout.work_dir(target_subnet), + keep_downloaded_data: self.params.keep_downloaded_state == Some(true), + data_includes: includes, + include_config: true, + }) + } + + fn validate_cup_step(&self, target_subnet: TargetSubnet) -> impl Step + use<> { + ValidateCupStep { + subnet_id: self.subnet_id(target_subnet), + target_subnet, + nns_url: self.recovery_args.nns_url.clone(), + layout: self.layout.clone(), + logger: self.recovery.logger.clone(), + } + } + + /// The recovery CUP of the destination subnet, at the merged state. + fn propose_cup(&self) -> RecoveryResult> { + let params = MergedStateParams::read(self.layout.merged_state_params_file())?; + + self.recovery.update_recovery_cup( + self.params.destination_subnet_id, + Height::from(params.height), + params.state_hash, + /*replacement_nodes=*/ &[], + /*registry_params=*/ None, + // The merged subnets are both unavailable while the recovery CUP is + // created, so its DKG is handled by whichever subnet the NNS picks + // by default, which is neither of them. + /*initial_dkg_subnet_id=*/ + None, + /*chain_key_subnet_id=*/ None, + Some(UNIX_EPOCH + Duration::from_nanos(params.time_nanos)), + ) + } + + fn upload_state_and_restart_step(&self) -> RecoveryResult> { + match self.params.upload_node_destination { + // The replica of the destination subnet has been stopped before its + // state was downloaded, so this step replaces its state directory + // and starts it back up on the merged state. + Some(node_ip) => Ok(UploadStateAndRestartStep { + logger: self.recovery.logger.clone(), + ssh_user: SshUser::Admin, + upload_method: DataLocation::Remote(node_ip), + work_dir: self.layout.work_dir(TargetSubnet::Merged), + data_src: self.layout.ic_state_dir(TargetSubnet::Merged), + require_confirmation: !self.recovery_args.skip_prompts, + key_file: self.recovery.admin_key_file.clone(), + check_ic_replay_height: false, + }), + None => Err(RecoveryError::StepSkipped), + } + } + + fn wait_for_recovery_cup_step(&self) -> RecoveryResult> { + match self.params.upload_node_destination { + Some(node_ip) => Ok(WaitForRecoveryCupStep { + node_ip, + layout: self.layout.clone(), + logger: self.recovery.logger.clone(), + }), + None => Err(RecoveryError::StepSkipped), + } + } + + fn ssh_helper(&self, node_ip: IpAddr) -> SshHelper { + let (ssh_user, key_file) = if self.params.readonly_pub_key.is_some() { + (SshUser::Readonly, self.params.readonly_key_file.clone()) + } else { + (SshUser::Admin, self.recovery.admin_key_file.clone()) + }; + + SshHelper::new( + self.recovery.logger.clone(), + ssh_user, + node_ip, + self.recovery.ssh_confirmation, + key_file, + ) + } + + fn download_node(&self, target_subnet: TargetSubnet) -> Option { + match target_subnet { + TargetSubnet::Source => self.params.download_node_source, + TargetSubnet::Destination => self.params.download_node_destination, + TargetSubnet::Merged => None, + } + } + + fn subnet_id(&self, target_subnet: TargetSubnet) -> SubnetId { + match target_subnet { + TargetSubnet::Source => self.params.source_subnet_id, + TargetSubnet::Destination | TargetSubnet::Merged => self.params.destination_subnet_id, + } + } + + fn dashboard_url(&self, dashboard: &str) -> String { + match self.recovery.registry_helper.latest_registry_version() { + Ok(registry_version) => format!( + "https://grafana.mainnet.dfinity.network/d/{dashboard}?var-datasource=IC+Metrics&var-ic=mercury&var-ic_subnet={}&var-registry_version={}", + self.params.source_subnet_id, registry_version, + ), + Err(err) => { + warn!( + self.logger, + "Failed to get the latest registry version: {}", err + ); + format!( + "https://grafana.mainnet.dfinity.network/d/{dashboard}?var-datasource=IC+Metrics&var-ic=mercury&var-ic_subnet={}", + self.params.source_subnet_id, + ) + } + } + } +} + +impl RecoveryIterator for SubnetMerging { + fn get_step_iterator(&mut self) -> &mut Peekable { + &mut self.step_iterator + } + + fn store_next_step(&mut self, step_type: Option) { + self.params.next_step = step_type; + } + + fn get_logger(&self) -> &Logger { + &self.logger + } + + fn interactive(&self) -> bool { + !self.recovery_args.skip_prompts + } + + fn read_step_params(&mut self, step_type: StepType) { + match step_type { + StepType::CoolDownSourceSubnet => { + if self.params.readonly_pub_key.is_none() { + self.params.readonly_pub_key = read_optional( + &self.logger, + "Enter public key to add readonly SSH access to both subnets. Ensure the right format.\n\ + Format: ssh-ed25519 \n\ + Example: ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIPwS/0S6xH0g/xLDV0Tz7VeMZE9AKPeSbLmCsq9bY3F1 foo@dfinity.org\n\ + Enter your key: ", + ) + } + } + + StepType::CheckMergeReadiness => { + print_url_and_ask_for_confirmation( + &self.logger, + self.dashboard_url("subnet-merging"), + "The tool evaluates the merge readiness condition itself, and waits until it \ + holds. Please check the dashboard as well, to see whether it is safe to \ + merge the subnet", + ); + } + + StepType::DownloadStateFromSourceSubnet => { + if self.params.download_node_source.is_none() { + self.params.download_node_source = read_optional( + &self.logger, + "Enter the IP of the node of the Source Subnet to download the state from:", + ); + } + + self.params.keep_downloaded_state = Some(consent_given( + &self.logger, + "Preserve original downloaded state locally?", + )); + } + + StepType::DownloadStateFromDestinationSubnet => { + if self.params.download_node_destination.is_none() { + self.params.download_node_destination = read_optional( + &self.logger, + "Enter the IP of the node of the Destination Subnet to download the state from:", + ); + } + } + + #[allow(clippy::collapsible_match)] + StepType::UploadStateToDestinationSubnet => { + if self.params.upload_node_destination.is_none() { + self.params.upload_node_destination = read_optional( + &self.logger, + "Enter IP of node in the Destination Subnet with admin access: ", + ); + } + } + + StepType::DeleteSourceSubnet => { + print_url_and_ask_for_confirmation( + &self.logger, + self.dashboard_url("subnet-merging"), + "Please check the dashboard to see if it is safe to delete the subnet that \ + was merged away", + ); + } + + _ => (), + } + } + + fn get_step_impl(&self, step_type: StepType) -> RecoveryResult> { + let step: Box = match step_type { + StepType::CoolDownSourceSubnet => AdminStep { + logger: self.recovery.logger.clone(), + ic_admin_cmd: get_propose_to_cool_down_subnet_command( + &self.recovery.admin_helper, + self.params.source_subnet_id, + &self.params.readonly_pub_key, + ), + } + .into(), + + StepType::CheckRegistryForCoolingDownFlag => CheckCoolingDownStep { + source_subnet_id: self.params.source_subnet_id, + registry_helper: self.recovery.registry_helper.clone(), + layout: self.layout.clone(), + timeout: Duration::from_secs(self.params.registry_timeout_secs), + poll_interval: Duration::from_secs(self.params.poll_interval_secs), + logger: self.recovery.logger.clone(), + } + .into(), + + StepType::CheckMergeReadiness => CheckMergeReadinessStep { + source_subnet_id: self.params.source_subnet_id, + registry_helper: self.recovery.registry_helper.clone(), + layout: self.layout.clone(), + timeout: Duration::from_secs(self.params.merge_ready_timeout_secs), + poll_interval: Duration::from_secs(self.params.poll_interval_secs), + logger: self.recovery.logger.clone(), + } + .into(), + + StepType::HaltSourceSubnetAtCupHeight => AdminStep { + logger: self.recovery.logger.clone(), + ic_admin_cmd: get_halt_subnet_at_cup_height_command( + &self.recovery.admin_helper, + self.params.source_subnet_id, + &self.params.readonly_pub_key, + ), + } + .into(), + + StepType::HaltDestinationSubnetAtCupHeight => AdminStep { + logger: self.recovery.logger.clone(), + ic_admin_cmd: get_halt_subnet_at_cup_height_command( + &self.recovery.admin_helper, + self.params.destination_subnet_id, + &self.params.readonly_pub_key, + ), + } + .into(), + + StepType::WaitForHaltingCupOnSourceSubnet => { + self.wait_for_halting_cup_step(TargetSubnet::Source)?.into() + } + StepType::WaitForHaltingCupOnDestinationSubnet => self + .wait_for_halting_cup_step(TargetSubnet::Destination)? + .into(), + + StepType::StopSourceReplica => self.stop_replica_step(TargetSubnet::Source)?.into(), + StepType::StopDestinationReplica => { + self.stop_replica_step(TargetSubnet::Destination)?.into() + } + + StepType::DownloadStateFromSourceSubnet => { + self.download_state_step(TargetSubnet::Source)?.into() + } + StepType::DownloadStateFromDestinationSubnet => { + self.download_state_step(TargetSubnet::Destination)?.into() + } + + StepType::ValidateSourceSubnetCup => { + self.validate_cup_step(TargetSubnet::Source).into() + } + StepType::ValidateDestinationSubnetCup => { + self.validate_cup_step(TargetSubnet::Destination).into() + } + + StepType::MergeStates => MergeStatesStep { + layout: self.layout.clone(), + time_margin: Duration::from_secs(self.params.time_margin_secs), + logger: self.recovery.logger.clone(), + } + .into(), + + StepType::MergeSubnets => AdminStep { + logger: self.recovery.logger.clone(), + ic_admin_cmd: get_propose_to_merge_subnets_command( + &self.recovery.admin_helper, + self.params.source_subnet_id, + self.params.destination_subnet_id, + ), + } + .into(), + + StepType::CheckRegistryForRoutingTableEntry => CheckRoutingTableStep { + source_subnet_id: self.params.source_subnet_id, + destination_subnet_id: self.params.destination_subnet_id, + registry_helper: self.recovery.registry_helper.clone(), + layout: self.layout.clone(), + timeout: Duration::from_secs(self.params.registry_timeout_secs), + poll_interval: Duration::from_secs(self.params.poll_interval_secs), + logger: self.recovery.logger.clone(), + } + .into(), + + StepType::ProposeCupForDestinationSubnet => self.propose_cup()?.into(), + StepType::UploadStateToDestinationSubnet => { + self.upload_state_and_restart_step()?.into() + } + StepType::WaitForCUPOnDestinationSubnet => self.wait_for_recovery_cup_step()?.into(), + + StepType::UnhaltDestinationSubnet => self + .recovery + .bring_subnet_back_online_after_repairs(self.params.destination_subnet_id) + .into(), + + StepType::CheckRegistryVersionOnAllSubnets => CheckRegistryVersionOnAllSubnetsStep { + source_subnet_id: self.params.source_subnet_id, + registry_helper: self.recovery.registry_helper.clone(), + layout: self.layout.clone(), + timeout: Duration::from_secs(self.params.merge_ready_timeout_secs), + poll_interval: Duration::from_secs(self.params.poll_interval_secs), + logger: self.recovery.logger.clone(), + } + .into(), + + StepType::DeleteSourceSubnet => AdminStep { + logger: self.recovery.logger.clone(), + ic_admin_cmd: get_propose_to_delete_subnet_command( + &self.recovery.admin_helper, + self.params.source_subnet_id, + ), + } + .into(), + + StepType::Cleanup => self.recovery.get_cleanup_step().into(), + }; + + Ok(step) + } +} + +impl Iterator for SubnetMerging { + type Item = (StepType, Box); + fn next(&mut self) -> Option { + self.next_step() + } +} + +impl HasRecoveryState for SubnetMerging { + type StepType = StepType; + type SubcommandArgsType = SubnetMergingArgs; + + fn get_next_step(&self) -> Option { + self.params.next_step + } + + fn get_state(&self) -> RecoveryResult> { + Ok(RecoveryState { + recovery_args: self.recovery_args.clone(), + neuron_args: self.neuron_args.clone(), + subcommand_args: self.params.clone(), + }) + } +} diff --git a/rs/recovery/subnet_merging/src/target_subnet.rs b/rs/recovery/subnet_merging/src/target_subnet.rs new file mode 100644 index 000000000000..b4fc210963b5 --- /dev/null +++ b/rs/recovery/subnet_merging/src/target_subnet.rs @@ -0,0 +1,25 @@ +/// The three states a subnet merge works with: the two states that are +/// downloaded and the merged one that is assembled from them. +#[derive(Copy, Clone, PartialEq, Debug)] +pub(crate) enum TargetSubnet { + /// The subnet that is cooling down, whose canisters are merged away and + /// which is deleted afterwards. + Source, + /// The subnet that hosts the canisters of the source subnet after the merge + /// and that is recovered at the merged state. + Destination, + /// Not a subnet of its own: the state assembled from the states of the two + /// subnets above, which the destination subnet is recovered at. + Merged, +} + +impl std::fmt::Display for TargetSubnet { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + let name = match self { + TargetSubnet::Source => "source", + TargetSubnet::Destination => "destination", + TargetSubnet::Merged => "merged", + }; + write!(f, "{name}") + } +} diff --git a/rs/recovery/subnet_merging/src/utils.rs b/rs/recovery/subnet_merging/src/utils.rs new file mode 100644 index 000000000000..fc43ad40b9a0 --- /dev/null +++ b/rs/recovery/subnet_merging/src/utils.rs @@ -0,0 +1,142 @@ +use ic_base_types::RegistryVersion; +use ic_recovery::{ + RECOVERY_DIRECTORY_NAME, + error::{RecoveryError, RecoveryResult}, + file_sync_helper::{read_file, write_file}, + registry_helper::RegistryHelper, +}; +use serde::{Deserialize, Serialize}; + +use std::path::Path; + +/// Everything the recovery of the destination subnet needs to know about the +/// merged state, as computed by the step that assembles it. +/// +/// Persisted next to the merged state so that the steps proposing the recovery +/// CUP and waiting for it agree on the same values, and so that a resumed run +/// neither re-merges the states nor re-hashes the merged checkpoint. +#[derive(Clone, PartialEq, Debug, Deserialize, Serialize)] +pub struct MergedStateParams { + /// The height of the merged checkpoint, i.e. the height the destination + /// subnet is recovered at. + pub height: u64, + /// The block time the recovered destination subnet starts from, in + /// nanoseconds since the Epoch. Larger than the batch times of both + /// checkpoints the merged state was assembled from. + pub time_nanos: u64, + /// The root hash of the manifest of the merged state. + pub state_hash: String, +} + +impl MergedStateParams { + pub fn read(path: &Path) -> RecoveryResult { + let contents = read_file(path)?; + serde_json::from_str(&contents).map_err(|err| { + RecoveryError::UnexpectedError(format!( + "Failed to parse the merged state params at {}: {err}", + path.display() + )) + }) + } + + pub fn write(&self, path: &Path) -> RecoveryResult<()> { + let contents = serde_json::to_string_pretty(self).map_err(|err| { + RecoveryError::UnexpectedError(format!( + "Failed to serialize the merged state params: {err}" + )) + })?; + write_file(path, contents) + } +} + +/// The lowest registry version at which `predicate` holds, in a registry whose +/// history has it hold from some version onwards -- e.g. "the subnet is labeled +/// cooling down" or "the subnet hosts no canister id range anymore", both of +/// which a single proposal of this tool brings about and nothing here undoes. +/// +/// Found by binary search, i.e. in a logarithmic number of (local) registry +/// reads rather than by scanning the whole history. Fails if the predicate does +/// not hold at the latest registry version; if the property was turned on and +/// off repeatedly, the version returned is the start of one of the stretches it +/// held in, which is why the callers log it for the operator to confirm. +pub(crate) fn first_registry_version_where( + registry_helper: &RegistryHelper, + predicate: impl Fn(RegistryVersion) -> RecoveryResult, +) -> RecoveryResult { + let latest = registry_helper.latest_registry_version()?; + if !predicate(latest)? { + return Err(RecoveryError::ValidationFailed(format!( + "The expected registry state has not been reached at the latest registry version \ + {latest}" + ))); + } + + // Invariant: the predicate does not hold at `low` and holds at `high`. The + // registry starts at version 1, whose predecessor 0 is the empty registry, + // where nothing holds. + let mut low = RegistryVersion::from(0); + let mut high = latest; + while high > low + RegistryVersion::from(1) { + let middle = RegistryVersion::from((low.get() + high.get()) / 2); + if predicate(middle)? { + high = middle; + } else { + low = middle; + } + } + + Ok(high) +} + +/// The file the tool records the registry version at which the subnet that is +/// merged away was labeled "cooling down" in, i.e. the `V` of the merge +/// readiness condition. +pub const COOLING_DOWN_REGISTRY_VERSION_FILE: &str = "cooling_down_registry_version"; +/// The file the tool records the registry version the merge was applied at in. +pub const MERGE_REGISTRY_VERSION_FILE: &str = "merge_registry_version"; + +pub fn write_registry_version(path: &Path, version: u64) -> RecoveryResult<()> { + write_file(path, version.to_string()) +} + +pub fn read_registry_version(path: &Path) -> RecoveryResult { + let contents = read_file(path)?; + contents.trim().parse().map_err(|err| { + RecoveryError::UnexpectedError(format!( + "Failed to parse the registry version {contents:?} at {}: {err}", + path.display() + )) + }) +} + +/// The registry version at which the subnet that is merged away was labeled +/// "cooling down", as recorded by the tool working in `dir`, i.e. in the +/// directory that was passed to it as `RecoveryArgs::dir`. +pub fn read_cooling_down_registry_version(dir: &Path) -> RecoveryResult { + read_registry_version( + &dir.join(RECOVERY_DIRECTORY_NAME) + .join(COOLING_DOWN_REGISTRY_VERSION_FILE), + ) +} + +#[cfg(test)] +mod tests { + use super::*; + + use ic_test_utilities_tmpdir::tmpdir; + + #[test] + fn merged_state_params_round_trip_test() { + let dir = tmpdir("test_dir"); + let path = dir.as_ref().join("merged_state_params.json"); + + let params = MergedStateParams { + height: 42_000, + time_nanos: 1_700_000_000_000_000_000, + state_hash: "deadbeef".to_string(), + }; + params.write(&path).unwrap(); + + assert_eq!(MergedStateParams::read(&path).unwrap(), params); + } +} diff --git a/rs/recovery/subnet_splitting/BUILD.bazel b/rs/recovery/subnet_splitting/BUILD.bazel index 70ee217a13cf..29ece9b3c6ea 100644 --- a/rs/recovery/subnet_splitting/BUILD.bazel +++ b/rs/recovery/subnet_splitting/BUILD.bazel @@ -13,11 +13,10 @@ rust_library( visibility = ["//rs:system-tests-pkg"], deps = [ # Keep sorted. - "//rs/crypto/utils/threshold_sig", - "//rs/crypto/utils/threshold_sig_der", "//rs/monitoring/metrics", "//rs/protobuf", "//rs/recovery", + "//rs/recovery/subnet_tools", "//rs/registry/routing_table", "//rs/registry/subnet_type", "//rs/state_layout", @@ -29,10 +28,7 @@ rust_library( "@crate_index//:anyhow", "@crate_index//:clap", "@crate_index//:csv", - "@crate_index//:hex", - "@crate_index//:ic-agent", "@crate_index//:serde", - "@crate_index//:serde_cbor", "@crate_index//:slog", "@crate_index//:strum", "@crate_index//:url", @@ -47,11 +43,10 @@ rust_binary( # Keep sorted. ":subnet_splitting", "//rs/canister_sandbox:backend_lib", - "//rs/crypto/utils/threshold_sig", - "//rs/crypto/utils/threshold_sig_der", "//rs/monitoring/metrics", "//rs/protobuf", "//rs/recovery", + "//rs/recovery/subnet_tools", "//rs/registry/routing_table", "//rs/registry/subnet_type", "//rs/state_layout", @@ -63,10 +58,7 @@ rust_binary( "@crate_index//:anyhow", "@crate_index//:clap", "@crate_index//:csv", - "@crate_index//:hex", - "@crate_index//:ic-agent", "@crate_index//:serde", - "@crate_index//:serde_cbor", "@crate_index//:slog", "@crate_index//:strum", "@crate_index//:url", @@ -79,11 +71,10 @@ rust_test( crate = "subnet_splitting", deps = [ # Keep sorted. - "//rs/crypto/utils/threshold_sig", - "//rs/crypto/utils/threshold_sig_der", "//rs/monitoring/metrics", "//rs/protobuf", "//rs/recovery", + "//rs/recovery/subnet_tools", "//rs/registry/routing_table", "//rs/registry/subnet_type", "//rs/state_layout", @@ -96,10 +87,7 @@ rust_test( "@crate_index//:anyhow", "@crate_index//:clap", "@crate_index//:csv", - "@crate_index//:hex", - "@crate_index//:ic-agent", "@crate_index//:serde", - "@crate_index//:serde_cbor", "@crate_index//:slog", "@crate_index//:strum", "@crate_index//:url", @@ -126,11 +114,10 @@ rust_ic_test( deps = [ # Keep sorted. ":subnet_splitting", - "//rs/crypto/utils/threshold_sig", - "//rs/crypto/utils/threshold_sig_der", "//rs/monitoring/metrics", "//rs/protobuf", "//rs/recovery", + "//rs/recovery/subnet_tools", "//rs/registry/routing_table", "//rs/registry/subnet_type", "//rs/rust_canisters/proxy_canister:lib", @@ -150,10 +137,7 @@ rust_ic_test( "@crate_index//:candid", "@crate_index//:clap", "@crate_index//:csv", - "@crate_index//:hex", - "@crate_index//:ic-agent", "@crate_index//:serde", - "@crate_index//:serde_cbor", "@crate_index//:slog", "@crate_index//:strum", "@crate_index//:url", diff --git a/rs/recovery/subnet_splitting/Cargo.toml b/rs/recovery/subnet_splitting/Cargo.toml index d1b30e6dc680..888a0e975ba4 100644 --- a/rs/recovery/subnet_splitting/Cargo.toml +++ b/rs/recovery/subnet_splitting/Cargo.toml @@ -10,11 +10,7 @@ documentation.workspace = true anyhow = { workspace = true } clap = { workspace = true } csv = { workspace = true } -hex = { workspace = true } -ic-agent = { workspace = true } ic-base-types = { path = "../../types/base_types/" } -ic-crypto-utils-threshold-sig = { path = "../../crypto/utils/threshold_sig" } -ic-crypto-utils-threshold-sig-der = { path = "../../crypto/utils/threshold_sig_der" } ic-metrics = { path = "../../monitoring/metrics" } ic-protobuf = { path = "../../protobuf" } ic-recovery = { path = "../" } @@ -23,10 +19,10 @@ ic-registry-subnet-type = { path = "../../registry/subnet_type" } ic-state-layout = { path = "../../state_layout" } ic-state-manager = { path = "../../state_manager" } ic-state-tool = { path = "../../state_tool" } +ic-subnet-tools = { path = "../subnet_tools" } ic-types-cycles = { path = "../../types/cycles" } ic-types = { path = "../../types/types" } serde = { workspace = true } -serde_cbor = { workspace = true } slog = { workspace = true } strum = { workspace = true } strum_macros = { workspace = true } diff --git a/rs/recovery/subnet_splitting/src/admin_helper.rs b/rs/recovery/subnet_splitting/src/admin_helper.rs index ceb676503d77..7a5d5c7bba5b 100644 --- a/rs/recovery/subnet_splitting/src/admin_helper.rs +++ b/rs/recovery/subnet_splitting/src/admin_helper.rs @@ -1,16 +1,12 @@ use crate::utils::canister_id_range_to_string; use ic_base_types::SubnetId; -use ic_recovery::admin_helper::{ - AdminHelper, CommandHelper, IcAdmin, SSH_READONLY_ACCESS_ARG, SUMMARY_ARG, quote, -}; +use ic_recovery::admin_helper::{AdminHelper, CommandHelper, IcAdmin, SUMMARY_ARG, quote}; use ic_registry_routing_table::CanisterIdRange; +use ic_subnet_tools::admin_helper::{DESTINATION_SUBNET_ARG, SOURCE_SUBNET_ARG}; -const SOURCE_SUBNET_ARG: &str = "source-subnet"; -const DESTINATION_SUBNET_ARG: &str = "destination-subnet"; const CANISTER_ID_RANGES_ARG: &str = "canister-id-ranges"; const MIGRATION_TRACE_ARG: &str = "migration-trace"; -const SUBNET_ARG: &str = "subnet"; /// Propose additions or updates to `canister_migrations`. /// @@ -92,36 +88,6 @@ pub(crate) fn get_propose_to_complete_canister_migration_command( ic_admin } -/// Propose to make the Subnet halt after reaching the next CUP height. -/// -/// Optionally adds a ssh-readonly-access key to the Subnet. -pub(crate) fn get_halt_subnet_at_cup_height_command( - admin_helper: &AdminHelper, - subnet_id: SubnetId, - key: &Option, -) -> IcAdmin { - let mut ic_admin = admin_helper.get_ic_admin_cmd_base(); - - ic_admin - .add_positional_argument("propose-to-update-subnet") - .add_argument(SUBNET_ARG, subnet_id) - .add_argument( - SUMMARY_ARG, - quote(format!( - "Halt subnet {subnet_id} at cup height and optionally update ssh readonly access", - )), - ) - .add_argument("halt-at-cup-height", true); - - if let Some(key) = key { - ic_admin.add_argument(SSH_READONLY_ACCESS_ARG, quote(key)); - } - - admin_helper.add_proposer_args(&mut ic_admin); - - ic_admin -} - #[cfg(test)] mod tests { use super::*; @@ -141,29 +107,6 @@ mod tests { "53zcu-tiaaa-aaaaa-qaaba-cai:54yea-6qaaa-aaaaa-qaabq-cai", "5h5yf-eiaaa-aaaaa-qaada-cai:5a46r-jqaaa-aaaaa-qaadq-cai", ]; - const SSH_KEY: &str = "fake ssh key"; - - #[test] - fn get_halt_subnet_at_cup_height_command_test() { - let result = get_halt_subnet_at_cup_height_command( - &fake_admin_helper(), - subnet_id_from_str(FAKE_SUBNET_ID_1), - &Some(SSH_KEY.to_string()), - ) - .join(" "); - - assert_eq!( - result, - "/fake/ic/admin/dir/ic-admin \ - --nns-url \"https://fake_nns_url.com:8080/\" \ - propose-to-update-subnet \ - --subnet gpvux-2ejnk-3hgmh-cegwf-iekfc-b7rzs-hrvep-5euo2-3ywz3-k3hcb-cqe \ - --summary \"Halt subnet gpvux-2ejnk-3hgmh-cegwf-iekfc-b7rzs-hrvep-5euo2-3ywz3-k3hcb-cqe at cup height and optionally update ssh readonly access\" \ - --halt-at-cup-height true \ - --ssh-readonly-access \"fake ssh key\" \ - --test-neuron-proposer" - ); - } #[test] fn get_propose_to_prepare_canister_migration_command_test() { diff --git a/rs/recovery/subnet_splitting/src/lib.rs b/rs/recovery/subnet_splitting/src/lib.rs index 1ebc3f829f89..d05c60727b7b 100644 --- a/rs/recovery/subnet_splitting/src/lib.rs +++ b/rs/recovery/subnet_splitting/src/lib.rs @@ -1,11 +1,8 @@ pub mod post_split_estimations; pub mod subnet_splitting; pub mod utils; -pub mod validation; mod admin_helper; -mod agent_helper; mod layout; -mod state_tool_helper; mod steps; mod target_subnet; diff --git a/rs/recovery/subnet_splitting/src/main.rs b/rs/recovery/subnet_splitting/src/main.rs index 45356a693fe7..c018241c199e 100644 --- a/rs/recovery/subnet_splitting/src/main.rs +++ b/rs/recovery/subnet_splitting/src/main.rs @@ -7,8 +7,8 @@ use ic_subnet_splitting::{ post_split_estimations, subnet_splitting::{SubnetSplitting, SubnetSplittingArgs}, utils::canister_id_ranges_to_strings, - validation::validate_artifacts, }; +use ic_subnet_tools::validation::validate_artifacts; use ic_types::ReplicaVersion; use slog::{Logger, info, warn}; use url::Url; diff --git a/rs/recovery/subnet_splitting/src/steps.rs b/rs/recovery/subnet_splitting/src/steps.rs index f8e51a41c5cb..a36258dd5f8e 100644 --- a/rs/recovery/subnet_splitting/src/steps.rs +++ b/rs/recovery/subnet_splitting/src/steps.rs @@ -1,28 +1,27 @@ use crate::{ - agent_helper::AgentHelper, - layout::Layout, - state_tool_helper, - target_subnet::TargetSubnet, - utils::{find_expected_state_hash_for_subnet_id, get_batch_time_from_cup, get_state_hash}, - validation::validate_artifacts, + layout::Layout, target_subnet::TargetSubnet, utils::find_expected_state_hash_for_subnet_id, }; use ic_base_types::SubnetId; use ic_metrics::MetricsRegistry; use ic_recovery::{ Recovery, - cli::consent_given, error::{RecoveryError, RecoveryResult}, file_sync_helper::rsync_includes, - registry_helper::VersionedRecoveryResult, steps::Step, util::parse_hex_str, }; use ic_registry_routing_table::CanisterIdRange; use ic_registry_subnet_type::SubnetType; use ic_state_manager::split::resolve_ranges_and_split; +use ic_subnet_tools::{ + agent_helper::AgentHelper, + state_tool_helper, + utils::{get_batch_time_from_cup, get_state_hash}, + validation::validate_artifacts, +}; use ic_types::Height; -use slog::{Logger, error, info}; +use slog::{Logger, info}; use url::Url; use std::{net::IpAddr, path::PathBuf}; @@ -283,34 +282,3 @@ impl Step for WaitForCUPStep { Recovery::wait_for_recovery_cup(&self.logger, self.node_ip, new_cup_height, state_hash) } } - -pub(crate) struct ReadRegistryStep VersionedRecoveryResult> { - pub(crate) logger: Logger, - pub(crate) label: String, - pub(crate) interactive: bool, - pub(crate) querier: F, -} - -impl VersionedRecoveryResult> Step for ReadRegistryStep { - fn descr(&self) -> String { - format!("Read Registry to get the most recent {}", self.label) - } - - fn exec(&self) -> RecoveryResult<()> { - loop { - match (self.querier)() { - Ok((registry_version, value)) => info!( - self.logger, - "{} at registry version {}: {:#?}", self.label, registry_version, value, - ), - Err(err) => error!(self.logger, "Failed getting {}, error: {}", self.label, err), - } - - if !self.interactive || !consent_given(&self.logger, "Read registry again?") { - break; - } - } - - Ok(()) - } -} diff --git a/rs/recovery/subnet_splitting/src/subnet_splitting.rs b/rs/recovery/subnet_splitting/src/subnet_splitting.rs index ef042f375b1a..f9a0d63fbc02 100644 --- a/rs/recovery/subnet_splitting/src/subnet_splitting.rs +++ b/rs/recovery/subnet_splitting/src/subnet_splitting.rs @@ -1,16 +1,15 @@ use crate::{ admin_helper::{ - get_halt_subnet_at_cup_height_command, get_propose_to_complete_canister_migration_command, + get_propose_to_complete_canister_migration_command, get_propose_to_prepare_canister_migration_command, get_propose_to_reroute_canister_ranges_command, }, layout::Layout, steps::{ - ComputeExpectedManifestsStep, CopyWorkDirStep, ReadRegistryStep, SplitStateStep, - StateSplitStrategy, ValidateCUPStep, WaitForCUPStep, + ComputeExpectedManifestsStep, CopyWorkDirStep, SplitStateStep, StateSplitStrategy, + ValidateCUPStep, WaitForCUPStep, }, target_subnet::TargetSubnet, - utils::get_state_hash, }; use clap::Parser; @@ -18,7 +17,7 @@ use ic_base_types::SubnetId; use ic_protobuf::registry::subnet::v1::SubnetRecord; use ic_recovery::{ CUPS_DIR, IC_STATE_DIR, NeuronArgs, Recovery, RecoveryArgs, - cli::{consent_given, read_optional, wait_for_confirmation}, + cli::{consent_given, read_optional}, error::{RecoveryError, RecoveryResult}, get_available_nodes_heights_from_metrics, recovery_iterator::RecoveryIterator, @@ -30,12 +29,15 @@ use ic_recovery::{ }; use ic_registry_routing_table::{CanisterIdRange, RoutingTable}; use ic_registry_subnet_type::SubnetType; +use ic_subnet_tools::{ + admin_helper::get_halt_subnet_at_cup_height_command, cli::print_url_and_ask_for_confirmation, + steps::ReadRegistryStep, utils::get_state_hash, +}; use ic_types::Height; use serde::{Deserialize, Serialize}; -use slog::{Logger, error, warn}; +use slog::{Logger, warn}; use strum::{EnumMessage, IntoEnumIterator}; use strum_macros::{EnumIter, EnumString}; -use url::Url; use std::{collections::HashMap, iter::Peekable, net::IpAddr, path::PathBuf}; @@ -328,6 +330,7 @@ impl SubnetSplitting { /*registry_params=*/ None, /*initial_dkg_subnet_id=*/ None, /*chain_key_subnet_id=*/ None, + /*time=*/ None, ) } @@ -690,20 +693,3 @@ impl HasRecoveryState for SubnetSplitting { }) } } - -fn print_url_and_ask_for_confirmation( - logger: &Logger, - url: String, - text_to_display: impl std::fmt::Display, -) { - match Url::parse(&url) { - Ok(url) => { - warn!(logger, "{}", text_to_display); - warn!(logger, "{}", url); - wait_for_confirmation(logger); - } - Err(err) => { - error!(logger, "Failed to parse url {}: {}", url, err); - } - } -} diff --git a/rs/recovery/subnet_splitting/src/utils.rs b/rs/recovery/subnet_splitting/src/utils.rs index f73b707647f0..c416b86b18b5 100644 --- a/rs/recovery/subnet_splitting/src/utils.rs +++ b/rs/recovery/subnet_splitting/src/utils.rs @@ -1,31 +1,12 @@ use ic_base_types::SubnetId; -use ic_protobuf::types::v1 as pb; use ic_recovery::{ error::{RecoveryError, RecoveryResult}, file_sync_helper::read_file, util::subnet_id_from_str, }; use ic_registry_routing_table::CanisterIdRange; -use ic_state_manager::manifest::{manifest_from_path, manifest_hash}; -use ic_types::{Time, consensus::CatchUpPackage}; -use std::{fmt::Display, path::Path}; - -pub(crate) fn get_batch_time_from_cup(cup_path: &Path) -> RecoveryResult