diff --git a/docs/operating/metastore-recovery.md b/docs/operating/metastore-recovery.md new file mode 100644 index 00000000000..607faa9d05f --- /dev/null +++ b/docs/operating/metastore-recovery.md @@ -0,0 +1,104 @@ +--- +title: Metastore recovery +sidebar_position: 5 +--- + +In case of disaster recovery you can rebuild the metastore state from +split recovery metadata. +The `quickwit tool recover` command reads that embedded metadata from each split and republishes +the split metadata into an existing index. + +## Requirements + +- **The index storage must support object listing.** Recovery discovers splits by listing the + index URI, which only S3-compatible storages implement. A `file://` index cannot be recovered + this way. +- **The splits must embed the recovery metadata.** It is written on every upload, but splits + produced by an older version of Quickwit do not have it and are skipped. +- **The index must exist, with the doc mapping the splits were built with.** Recovery restores + splits, not the index configuration. A mismatched doc mapping yields an index that is searchable + but returns wrong or missing fields for the recovered splits. + +## Recovering an index + +### 1. Recreate the index + +Create the index with the same doc mapping as before the loss, and make sure its `index_uri` +points at the directory holding the splits: + +```bash +curl -X POST http://localhost:7280/api/v1/indexes \ + -H "Content-Type: application/yaml" \ + --data-binary @index-config.yaml +``` + +### 2. Preview the recovery + +```bash +quickwit tool recover --index otel-logs-v0_9 --config ./config/quickwit.yaml --dry-run +``` + +``` +Recovering index otel-logs-v0_9 from s3://my-bucket/indexes/otel-logs-v0_9 + ++-------------------------------------+-------+ +| Recovery summary (dry run) | ++-------------------------------------+-------+ +| Splits found in storage | 12 | +| Publishable | 5 | +| Documents | 1,099 | +| Skipped (merged into another split) | 0 | +| Skipped (already in metastore) | 6 | +| Replaced (marked for deletion) | 2 | +| Unreadable | 1 | ++-------------------------------------+-------+ +``` + +The command then lists the splits it would publish, the IDs it skipped, and the split files it +could not read along with the reason. + +### 3. Publish the splits + +Run the same command without `--dry-run`. The splits are staged and published in batches. + +### 4. Verify + +Compare the hit count over the time range covered by the recovered splits with the document count +reported by the command: + +```bash +curl "http://localhost:7280/api/v1/otel-logs-v0_9/search?query=*&max_hits=0\ +&start_timestamp=1792257120&end_timestamp=1792258260" +``` + +## What the command does + +1. Lists the `.split` objects at the root of the index URI. +2. Fetches the recovery metadata of each split, ten splits at a time. The object size comes from + the listing, and the footer and the metadata file usually land in a single `GET` per split. +3. Rebuilds the split metadata, replacing the index UID recorded in the split with the UID of the + target index. Splits belonging to another index are skipped. +4. Skips the splits that were superseded by a merge. A merged split and its parents coexist in the + storage until garbage collection removes the parents. +5. Leaves out the splits the metastore already knows about, which makes a second run a no-op. + Splits left staged by an interrupted run are not considered known: they are staged and + published again. +6. Stages and publishes the remaining splits, 1000 per request. A recovered merged split replaces + the splits it was merged out of, transitively: if the metastore still holds any of them as + published, for instance because it was restored from a backup predating the merge, they are + marked for deletion in the same operation, so that their documents are not returned twice. + +Splits whose metadata cannot be read are reported individually and do not interrupt the run: a +single truncated upload would otherwise make every other split unrecoverable. + +## What is not recovered + +- **Source checkpoints.** The index checkpoint is left empty. A source restarted against a + recovered index resumes from the position its own configuration dictates, which may re-ingest or + skip data depending on the source. +- **Delete tasks.** Neither the delete task queue nor the delete opstamps are restored: the + recovered splits are published with an opstamp of `0`, since there is no way to tell which + delete tasks were applied to them. Pending deletes have to be submitted again, and already + applied ones are not replayed. +- **Splits without embedded recovery metadata.** They are listed as unreadable and left in the + storage, untouched. diff --git a/quickwit/quickwit-cli/src/tool.rs b/quickwit/quickwit-cli/src/tool.rs index dfff0b937c2..5275948c1f6 100644 --- a/quickwit/quickwit-cli/src/tool.rs +++ b/quickwit/quickwit-cli/src/tool.rs @@ -22,11 +22,13 @@ use std::time::{Duration, Instant}; use std::{env, fmt, io}; use anyhow::{Context, bail}; +use bytesize::ByteSize; use clap::{ArgMatches, Command, arg}; use colored::{ColoredString, Colorize}; use humantime::format_duration; use quickwit_actors::{ActorExitStatus, ActorHandle, Mailbox, Universe}; use quickwit_cluster::{ChitchatTransport, Cluster, ClusterMember, FailureDetectorConfig}; +use quickwit_common::pretty::PrettySample; use quickwit_common::pubsub::EventBroker; use quickwit_common::runtimes::RuntimesConfig; use quickwit_common::uri::Uri; @@ -35,7 +37,9 @@ use quickwit_config::{ CLI_SOURCE_ID, IndexerConfig, NodeConfig, SourceConfig, SourceInputFormat, SourceParams, TransformConfig, VecSourceParams, }; -use quickwit_index_management::{IndexService, clear_cache_directory}; +use quickwit_index_management::{ + IndexService, SplitRecoveryInfo, SplitToPublish, clear_cache_directory, +}; use quickwit_indexing::actors::{IndexingService, MergePipeline, MergeSchedulerService}; use quickwit_indexing::docs_clustering::Fingerprinter; use quickwit_indexing::models::{ @@ -43,7 +47,7 @@ use quickwit_indexing::models::{ }; use quickwit_indexing::{IndexingPipeline, IndexingSplitCache}; use quickwit_ingest::IngesterPool; -use quickwit_metastore::IndexMetadataResponseExt; +use quickwit_metastore::{IndexMetadataResponseExt, SplitMetadata}; use quickwit_proto::indexing::CpuCapacity; use quickwit_proto::ingest::ingester::IngesterStatus; use quickwit_proto::metastore::{IndexMetadataRequest, MetastoreService, MetastoreServiceClient}; @@ -55,12 +59,15 @@ use quickwit_serve::{ }; use quickwit_storage::{BundleStorage, Storage}; use quickwit_transport::ChannelFactory; +use tabled::Tabled; use thousands::Separable; +use time::OffsetDateTime; +use time::format_description::well_known::Rfc3339; use tracing::debug; -use crate::checklist::{GREEN_COLOR, RED_COLOR}; +use crate::checklist::{GREEN_COLOR, RED_COLOR, WHITE_COLOR}; use crate::{ - THROUGHPUT_WINDOW_SIZE, config_cli_arg, get_resolvers, info, load_node_config, + THROUGHPUT_WINDOW_SIZE, config_cli_arg, get_resolvers, info, load_node_config, make_table, run_index_checklist, start_actor_runtimes, }; @@ -165,9 +172,25 @@ pub fn build_tool_command() -> Command { .required(true), ]) ) + .subcommand(Command::new("recover").display_order(11) + .about("Recovers an index from the metadata embedded in its splits. Splits written by a version of Quickwit that does not embed this metadata are skipped. Requires an existing index and an S3-compatible index storage.").args(&[ + arg!(--index "ID of the target index") + .display_order(1) + .required(true), + arg!(--"dry-run" "Executes the command in dry run mode and only displays the splits that are going to be added to the index") + .required(false), + ]) + ) .arg_required_else_help(true) } +#[derive(Debug, Eq, PartialEq)] +pub struct RecoverIndexArgs { + pub config_uri: Uri, + pub index_id: IndexId, + pub dry_run: bool, +} + #[derive(Debug, Eq, PartialEq)] pub struct LocalIngestDocsArgs { pub config_uri: Uri, @@ -223,6 +246,7 @@ pub enum ToolCliCommand { LocalIngest(LocalIngestDocsArgs), LocalSearch(LocalSearchArgs), ExtractSplit(ExtractSplitArgs), + RecoverIndex(RecoverIndexArgs), Merge(MergeArgs), } @@ -237,6 +261,7 @@ impl ToolCliCommand { "local-search" => Self::parse_local_search_args(submatches), "merge" => Self::parse_merge_args(submatches), "extract-split" => Self::parse_extract_split_args(submatches), + "recover" => Self::parse_recover_index_args(submatches), _ => bail!("unknown tool subcommand `{subcommand}`"), } } @@ -364,6 +389,23 @@ impl ToolCliCommand { })) } + fn parse_recover_index_args(mut matches: ArgMatches) -> anyhow::Result { + let config_uri = matches + .get_one("config") + .map(|uri_str: &String| Uri::from_str(uri_str)) + .expect("`config` should be a required arg.")?; + let index_id = matches + .remove_one::("index") + .expect("`index` should be a required arg."); + let dry_run = matches.get_flag("dry-run"); + + Ok(Self::RecoverIndex(RecoverIndexArgs { + config_uri, + index_id, + dry_run, + })) + } + fn parse_extract_split_args(mut matches: ArgMatches) -> anyhow::Result { let index_id = matches .remove_one::("index") @@ -395,6 +437,7 @@ impl ToolCliCommand { Self::LocalSearch(args) => local_search_cli(args).await, Self::Merge(args) => merge_cli(args).await, Self::ExtractSplit(args) => extract_split_cli(args).await, + Self::RecoverIndex(args) => recover_index_cli(args).await, } } } @@ -685,6 +728,217 @@ pub async fn garbage_collect_index_cli(args: GarbageCollectIndexArgs) -> anyhow: print_tantivy_gc_result(args.dry_run, removal_info) } +pub async fn recover_index_cli(args: RecoverIndexArgs) -> anyhow::Result<()> { + let config = load_node_config(&args.config_uri, None).await?; + let (storage_resolver, metastore_resolver) = + get_resolvers(&config.storage_configs, &config.metastore_configs); + let metastore = metastore_resolver.resolve(&config.metastore_uri).await?; + let mut index_service = IndexService::new(metastore.clone(), storage_resolver); + + println!("Recovering index {}\n", args.index_id.bold()); + let recovery_info = index_service + .recover_index_from_storage(&args.index_id, args.dry_run) + .await?; + + print_split_recovery_info(&args.index_id, &recovery_info); + Ok(()) +} + +#[derive(Tabled)] +struct RecoveredSplitRow { + #[tabled(rename = "Split ID")] + split_id: String, + #[tabled(rename = "Docs")] + num_docs: String, + #[tabled(rename = "Split size")] + split_size: String, + #[tabled(rename = "Time range")] + time_range: String, + #[tabled(rename = "Created at")] + created_at: String, + #[tabled(rename = "Merge ops")] + num_merge_ops: usize, + #[tabled(rename = "Doc mapping UID")] + doc_mapping_uid: String, + #[tabled(rename = "Replaces")] + num_replaced_splits: usize, +} + +impl From<&SplitToPublish> for RecoveredSplitRow { + fn from(split_to_publish: &SplitToPublish) -> Self { + let split_metadata = &split_to_publish.split_metadata; + // The footer ends at the very end of the split file, so its end offset is the file size. + let split_size = ByteSize::b(split_metadata.footer_offsets.end); + RecoveredSplitRow { + split_id: split_metadata.split_id().to_string(), + num_docs: split_metadata.num_docs.separate_with_commas(), + split_size: split_size.to_string(), + time_range: format_time_range(split_metadata), + created_at: format_timestamp(split_metadata.create_timestamp), + num_merge_ops: split_metadata.num_merge_ops, + doc_mapping_uid: split_metadata.doc_mapping_uid.to_string(), + num_replaced_splits: split_to_publish.replaced_split_ids.len(), + } + } +} + +fn format_timestamp(unix_timestamp: i64) -> String { + let Ok(date_time) = OffsetDateTime::from_unix_timestamp(unix_timestamp) else { + return unix_timestamp.to_string(); + }; + date_time + .format(&Rfc3339) + .unwrap_or_else(|_| unix_timestamp.to_string()) +} + +fn format_time_range(split_metadata: &SplitMetadata) -> String { + let Some(time_range) = &split_metadata.time_range else { + return "*".to_string(); + }; + format!( + "{} → {}", + format_timestamp(*time_range.start()), + format_timestamp(*time_range.end()) + ) +} + +#[derive(Tabled)] +struct SplitRecoverySummaryRow { + #[tabled(rename = "Splits found in storage")] + num_split_files: String, + #[tabled(rename = "Publishable")] + num_published: String, + #[tabled(rename = "Documents")] + num_docs_published: String, + #[tabled(rename = "Skipped (merged into another split)")] + num_merged: String, + #[tabled(rename = "Skipped (already in metastore)")] + num_already_known: String, + #[tabled(rename = "Replaced (marked for deletion)")] + num_replaced: String, + #[tabled(rename = "Unreadable")] + num_unreadable: String, +} + +fn print_split_recovery_info(index_id: &str, recovery_info: &SplitRecoveryInfo) { + let num_published = recovery_info.published_splits.len(); + let num_unreadable = recovery_info.unreadable_splits.len(); + + let summary_row = SplitRecoverySummaryRow { + num_split_files: recovery_info.num_split_files().separate_with_commas(), + num_published: num_published.separate_with_commas(), + num_docs_published: recovery_info.num_docs_published().separate_with_commas(), + num_merged: recovery_info.merged_split_ids.len().separate_with_commas(), + num_already_known: recovery_info + .already_known_split_ids + .len() + .separate_with_commas(), + num_replaced: recovery_info.num_replaced_splits().separate_with_commas(), + num_unreadable: num_unreadable.separate_with_commas(), + }; + let summary_header = if recovery_info.dry_run { + "Recovery summary (dry run)" + } else { + "Recovery summary" + }; + println!("{}", make_table(summary_header, [summary_row], true)); + + if !recovery_info.published_splits.is_empty() { + let splits_header = if recovery_info.dry_run { + "Splits that would be published" + } else { + "Published splits" + }; + let rows = recovery_info + .published_splits + .iter() + .map(RecoveredSplitRow::from); + println!("\n{}", make_table(splits_header, rows, false)); + } + print_skipped_split_ids( + "merged into another recovered split", + &recovery_info.merged_split_ids, + ); + print_skipped_split_ids( + "already known to the metastore", + &recovery_info.already_known_split_ids, + ); + print_replaced_split_ids(recovery_info); + + if num_unreadable > 0 { + println!( + "\n{} {} split file(s) could not be read:", + "✖".color(RED_COLOR), + num_unreadable + ); + for unreadable_split in &recovery_info.unreadable_splits { + println!( + " {} {}: {:#}", + "✖".color(RED_COLOR), + unreadable_split.split_file.display(), + unreadable_split.error + ); + } + } + + println!(); + if recovery_info.dry_run { + println!( + "{} Dry run: nothing was published. Re-run without `--dry-run` to publish {} split(s) \ + into `{index_id}`.", + "!".color(WHITE_COLOR), + num_published + ); + return; + } + if num_published == 0 { + println!( + "{} No split to publish into `{index_id}`.", + "✔".color(GREEN_COLOR) + ); + return; + } + println!( + "{} Published {} split(s) holding {} document(s) into `{index_id}`.", + "✔".color(GREEN_COLOR), + num_published.separate_with_commas(), + recovery_info.num_docs_published().separate_with_commas() + ); +} + +/// Prints the IDs of the published splits superseded by a recovered merge, sampled to keep the +/// output readable. +fn print_replaced_split_ids(recovery_info: &SplitRecoveryInfo) { + let replaced_split_ids: Vec<&SplitId> = recovery_info.replaced_split_ids().collect(); + if replaced_split_ids.is_empty() { + return; + } + let outcome = if recovery_info.dry_run { + "would be marked for deletion" + } else { + "marked for deletion" + }; + println!( + "\n{} {} split(s) {outcome}, replaced by a recovered merged split: {:?}", + "-".color(WHITE_COLOR), + replaced_split_ids.len(), + PrettySample::new(&replaced_split_ids, 10) + ); +} + +/// Prints the IDs of the splits left out of the publication, sampled to keep the output readable. +fn print_skipped_split_ids(reason: &str, split_ids: &[SplitId]) { + if split_ids.is_empty() { + return; + } + println!( + "\n{} {} split(s) skipped, {reason}: {:?}", + "-".color(WHITE_COLOR), + split_ids.len(), + PrettySample::new(split_ids, 10) + ); +} + fn print_tantivy_gc_result( dry_run: bool, removal_info: quickwit_index_management::SplitRemovalInfo, diff --git a/quickwit/quickwit-index-management/src/index.rs b/quickwit/quickwit-index-management/src/index.rs index 4c49bd3b5e6..cfea461a74b 100644 --- a/quickwit/quickwit-index-management/src/index.rs +++ b/quickwit/quickwit-index-management/src/index.rs @@ -13,33 +13,41 @@ // limitations under the License. use std::collections::{HashMap, HashSet}; -use std::path::Path; +use std::ffi::OsStr; +use std::path::{Path, PathBuf}; +use std::sync::Arc; use std::time::Duration; +use anyhow::Context; +use futures::TryStreamExt; use futures_util::StreamExt; use itertools::Itertools; use quickwit_common::fs::{empty_dir, get_cache_directory_path}; use quickwit_common::pretty::PrettySample; use quickwit_common::rate_limited_error; +use quickwit_common::shared_consts::SPLIT_RECOVERY_METADATA_FILE_NAME; use quickwit_config::{IndexConfig, SourceConfig, validate_identifier}; use quickwit_indexing::check_source_connectivity; use quickwit_metastore::{ AddSourceRequestExt, CreateIndexResponseExt, IndexMetadata, IndexMetadataResponseExt, ListIndexesMetadataResponseExt, ListSplitsQuery, ListSplitsRequestExt, - MetastoreServiceStreamSplitsExt, SplitInfo, SplitMetadata, SplitState, UpdateIndexRequestExt, - UpdateSourceRequestExt, + MetastoreServiceStreamSplitsExt, SplitInfo, SplitMetadata, SplitState, StageSplitsRequestExt, + UpdateIndexRequestExt, UpdateSourceRequestExt, }; use quickwit_proto::metastore::{ AddSourceRequest, CreateIndexRequest, DeleteIndexRequest, EntityKind, IndexMetadataRequest, ListIndexesMetadataRequest, ListSplitsRequest, MarkSplitsForDeletionRequest, MetastoreError, - MetastoreService, MetastoreServiceClient, ResetSourceCheckpointRequest, UpdateIndexRequest, - UpdateSourceRequest, serde_utils, + MetastoreService, MetastoreServiceClient, PublishSplitsRequest, ResetSourceCheckpointRequest, + SplitRecoveryMetadata, StageSplitsRequest, UpdateIndexRequest, UpdateSourceRequest, + serde_utils, }; use quickwit_proto::types::{IndexUid, SplitId}; use quickwit_proto::{ServiceError, ServiceErrorCode}; -use quickwit_storage::{StorageResolver, StorageResolverError}; +use quickwit_storage::{ + BundleStorage, ObjectMetadata, Storage, StorageResolver, StorageResolverError, +}; use thiserror::Error; -use tracing::{error, info}; +use tracing::{error, info, warn}; use crate::garbage_collection::{ DeleteSplitsError, SplitRemovalInfo, delete_splits_from_storage_and_metastore, @@ -563,6 +571,389 @@ impl IndexService { Ok(source_config) } + + pub async fn recover_index_from_storage( + &mut self, + index_id: &str, + dry_run: bool, + ) -> Result { + let index_metadata_request = IndexMetadataRequest::for_index_id(index_id.to_string()); + let index_metadata = self + .metastore + .index_metadata(index_metadata_request) + .await? + .deserialize_index_metadata()?; + + let storage = self + .storage_resolver + .resolve(&index_metadata.index_config().index_uri) + .await?; + + let read_outcome = read_splits_recovery_metadata(&storage).await?; + let index_uid = index_metadata.index_uid; + + if !read_outcome.unreadable_splits.is_empty() { + warn!( + index_id=%index_id, + num_unreadable_splits=read_outcome.unreadable_splits.len(), + "some split files could not be read" + ); + } + info!( + index_id=%index_id, + dry_run=dry_run, + num_recovered_splits=read_outcome.recovered_splits.len(), + num_unreadable_splits=read_outcome.unreadable_splits.len(), + "read split recovery metadata" + ); + + let known_split_states: HashMap = + self.list_metastore_split_states(index_uid.clone()).await?; + let plan = plan_split_recovery( + &index_uid, + read_outcome.recovered_splits, + &known_split_states, + ); + + let recovery_info = SplitRecoveryInfo { + dry_run, + published_splits: plan.splits_to_publish, + merged_split_ids: plan.merged_split_ids, + already_known_split_ids: plan.already_known_split_ids, + unreadable_splits: read_outcome.unreadable_splits, + }; + if dry_run { + return Ok(recovery_info); + } + + for splits_batch in recovery_info + .published_splits + .chunks(PUBLISH_SPLITS_BATCH_SIZE) + { + self.publish_recovered_splits(&index_uid, splits_batch) + .await?; + } + info!( + index_id=%index_id, + num_published_splits=recovery_info.published_splits.len(), + num_published_docs=recovery_info.num_docs_published(), + num_replaced_splits=recovery_info.num_replaced_splits(), + "published recovered splits" + ); + Ok(recovery_info) + } + + /// Lists the state of each split the metastore already holds as published or marked for + /// deletion. + async fn list_metastore_split_states( + &mut self, + index_uid: IndexUid, + ) -> Result, IndexServiceError> { + let list_splits_query = ListSplitsQuery::for_index(index_uid) + .with_split_states([SplitState::Published, SplitState::MarkedForDeletion]); + let list_splits_request = + ListSplitsRequest::try_from_list_splits_query(&list_splits_query)?; + let split_states: HashMap = self + .metastore + .list_splits(list_splits_request) + .await? + .collect_splits() + .await? + .into_iter() + .map(|split| (split.split_metadata.split_id, split.split_state)) + .collect(); + Ok(split_states) + } + + /// Stages then publishes a batch of recovered splits. + async fn publish_recovered_splits( + &mut self, + index_uid: &IndexUid, + splits_to_publish: &[SplitToPublish], + ) -> Result<(), IndexServiceError> { + let stage_splits_request = StageSplitsRequest::try_from_splits_metadata( + index_uid.clone(), + splits_to_publish + .iter() + .map(|split_to_publish| split_to_publish.split_metadata.clone()), + )?; + self.metastore.stage_splits(stage_splits_request).await?; + + let publish_splits_request = PublishSplitsRequest { + index_uid: Some(index_uid.clone()), + staged_split_ids: splits_to_publish + .iter() + .map(|split_to_publish| split_to_publish.split_metadata.split_id().to_string()) + .collect(), + replaced_split_ids: splits_to_publish + .iter() + .flat_map(|split_to_publish| split_to_publish.replaced_split_ids.iter()) + .map(|split_id| split_id.to_string()) + .collect(), + index_checkpoint_delta_json_opt: None, + publish_token_opt: None, + }; + self.metastore + .publish_splits(publish_splits_request) + .await?; + Ok(()) + } +} + +struct SplitRecoveryPlan { + splits_to_publish: Vec, + merged_split_ids: Vec, + already_known_split_ids: Vec, +} + +/// Decides which recovered splits should be published into `index_uid`. +/// +/// Ignores split ids already stored in the metastore and splits that got +/// merged together and that still remain (probably with a pending deletion before the disaster). +fn plan_split_recovery( + index_uid: &IndexUid, + recovered_splits: Vec, + known_split_states: &HashMap, +) -> SplitRecoveryPlan { + let parent_split_ids: HashMap<&SplitId, &[SplitId]> = recovered_splits + .iter() + .map(|recovered_split| { + ( + recovered_split.split_metadata.split_id(), + recovered_split.parent_split_ids.as_slice(), + ) + }) + .collect(); + let merged_split_ids: HashSet<&SplitId> = recovered_splits + .iter() + .flat_map(|recovered_split| recovered_split.parent_split_ids.iter()) + .collect(); + + let mut plan = SplitRecoveryPlan { + splits_to_publish: Vec::new(), + merged_split_ids: Vec::new(), + already_known_split_ids: Vec::new(), + }; + let mut visited_ancestor_ids: HashSet = HashSet::new(); + for recovered_split in &recovered_splits { + let split_id = recovered_split.split_metadata.split_id(); + if merged_split_ids.contains(split_id) { + plan.merged_split_ids.push(split_id.clone()); + continue; + } + if known_split_states.contains_key(split_id) { + plan.already_known_split_ids.push(split_id.clone()); + continue; + } + let recovered_index_uid = &recovered_split.split_metadata.index_uid; + if recovered_index_uid.index_id != index_uid.index_id { + warn!( + split_id = %recovered_split.split_metadata.split_id, + found_index_id = %recovered_index_uid.index_id, + expected_index_id = %index_uid.index_id, + "skipping split belonging to another index" + ); + continue; + } + let replaced_split_ids = collect_replaced_split_ids( + split_id, + &parent_split_ids, + known_split_states, + &mut visited_ancestor_ids, + ); + // we've just checked index_id is the same, we overwriting the uid to keep + // the incarnation id uniform + let mut split_metadata = recovered_split.split_metadata.clone(); + split_metadata.index_uid = index_uid.clone(); + // set opstamp to 0 as we dont know the state of delete tasks + split_metadata.delete_opstamp = 0; + plan.splits_to_publish.push(SplitToPublish { + split_metadata, + replaced_split_ids, + }); + } + plan +} + +/// Collects the ancestors of `split_id` in the merge graph that the metastore still holds as +/// published: publishing `split_id` must replace them, otherwise their documents would show up +/// twice. +fn collect_replaced_split_ids( + split_id: &SplitId, + parent_split_ids: &HashMap<&SplitId, &[SplitId]>, + known_split_states: &HashMap, + visited_ancestor_ids: &mut HashSet, +) -> Vec { + let mut replaced_split_ids: Vec = Vec::new(); + let mut ancestor_ids_to_visit: Vec<&SplitId> = parent_split_ids + .get(split_id) + .copied() + .unwrap_or_default() + .iter() + .collect(); + + while let Some(ancestor_id) = ancestor_ids_to_visit.pop() { + if !visited_ancestor_ids.insert(ancestor_id.clone()) { + continue; + } + if known_split_states.get(ancestor_id) == Some(&SplitState::Published) { + replaced_split_ids.push(ancestor_id.clone()); + } + // The ancestor may itself be the result of a merge: its own parents are superseded too. + if let Some(grand_parent_split_ids) = parent_split_ids.get(ancestor_id) { + ancestor_ids_to_visit.extend(grand_parent_split_ids.iter()); + } + } + replaced_split_ids +} + +const READ_RECOVERY_METADATA_CONCURRENCY: usize = 10; + +const PUBLISH_SPLITS_BATCH_SIZE: usize = 1_000; + +#[derive(Debug)] +struct RecoveredSplit { + split_metadata: SplitMetadata, + parent_split_ids: Vec, +} + +#[derive(Debug)] +pub struct UnreadableSplit { + pub split_file: PathBuf, + pub error: anyhow::Error, +} + +#[derive(Debug, Default)] +struct SplitsRecoveryMetadata { + pub recovered_splits: Vec, + pub unreadable_splits: Vec, +} + +#[derive(Debug)] +pub struct SplitToPublish { + pub split_metadata: SplitMetadata, + pub replaced_split_ids: Vec, +} + +pub struct SplitRecoveryInfo { + /// Whether the splits were only planned, and not actually published. + pub dry_run: bool, + /// Splits published into the index, or that a dry run would have published. + pub published_splits: Vec, + /// Recovered splits that was merged into another split. + pub merged_split_ids: Vec, + /// Recovered splits the metastore already knew about. + pub already_known_split_ids: Vec, + /// Split files whose recovery metadata could not be read. + pub unreadable_splits: Vec, +} + +impl SplitRecoveryInfo { + pub fn num_split_files(&self) -> usize { + self.published_splits.len() + + self.merged_split_ids.len() + + self.already_known_split_ids.len() + + self.unreadable_splits.len() + } + + pub fn num_docs_published(&self) -> u64 { + self.published_splits + .iter() + .map(|split_to_publish| split_to_publish.split_metadata.num_docs as u64) + .sum() + } + + /// Returns the ids of the published splits replaced by the recovered splits, and hence marked + /// for deletion. + pub fn replaced_split_ids(&self) -> impl Iterator { + self.published_splits + .iter() + .flat_map(|split_to_publish| split_to_publish.replaced_split_ids.iter()) + } + + pub fn num_replaced_splits(&self) -> usize { + self.replaced_split_ids().count() + } +} + +async fn read_splits_recovery_metadata( + storage: &Arc, +) -> Result { + let split_objects: Vec = list_split_objects(storage).await?; + + let read_results: Vec<(PathBuf, anyhow::Result)> = + futures::stream::iter(split_objects) + .map(|split_object| { + let storage = storage.clone(); + async move { + let split_file = split_object.path.clone(); + let read_result = read_split_recovery_metadata(storage, split_object).await; + (split_file, read_result) + } + }) + .buffer_unordered(READ_RECOVERY_METADATA_CONCURRENCY) + .collect() + .await; + + let mut outcome = SplitsRecoveryMetadata::default(); + for (split_file, read_result) in read_results { + match read_result { + Ok(recovered_split) => outcome.recovered_splits.push(recovered_split), + Err(error) => outcome + .unreadable_splits + .push(UnreadableSplit { split_file, error }), + } + } + Ok(outcome) +} + +async fn list_split_objects( + storage: &Arc, +) -> Result, IndexServiceError> { + let object_pages: Vec> = storage + .list(Path::new("")) + .try_collect() + .await + .map_err(|error| { + IndexServiceError::Internal(format!( + "failed to list objects in `{}`: {error}", + storage.uri() + )) + })?; + + let split_objects: Vec = object_pages + .into_iter() + .flatten() + .filter(|object| object.path.extension() == Some(OsStr::new("split"))) + .collect(); + + Ok(split_objects) +} + +async fn read_split_recovery_metadata( + storage: Arc, + split_object: ObjectMetadata, +) -> anyhow::Result { + let (recovery_metadata_bytes, footer_offsets) = BundleStorage::fetch_file_from_split( + storage, + PathBuf::from(SPLIT_RECOVERY_METADATA_FILE_NAME), + &split_object.path, + split_object.size.as_u64(), + ) + .await + .context("failed to fetch split recovery metadata")?; + + let recovery_metadata = SplitRecoveryMetadata::deserialize(&recovery_metadata_bytes) + .context("failed to deserialize split recovery metadata")?; + + let (split_metadata, parent_split_ids) = + SplitMetadata::try_from_recovery_metadata(recovery_metadata, footer_offsets) + .context("failed to rebuild split metadata")?; + + Ok(RecoveredSplit { + split_metadata, + parent_split_ids, + }) } /// Clears the cache directory of a given source. @@ -770,4 +1161,254 @@ mod tests { assert!(splits.is_empty()); assert!(!storage.exists(split_path).await.unwrap()); } + + #[tokio::test] + async fn test_list_metastore_split_states_ignores_staged_splits() { + let metastore = metastore_for_test(); + let storage_resolver = StorageResolver::for_test(); + let mut index_service = IndexService::new(metastore.clone(), storage_resolver); + let index_config = IndexConfig::for_test("test-index", "ram://indexes/test-index"); + let index_uid = index_service + .create_index(index_config, false) + .await + .unwrap() + .index_uid; + + let staged_split = SplitMetadata::for_test("staged-split".into()); + let published_split = SplitMetadata::for_test("published-split".into()); + let stage_splits_request = StageSplitsRequest::try_from_splits_metadata( + index_uid.clone(), + [staged_split, published_split], + ) + .unwrap(); + metastore.stage_splits(stage_splits_request).await.unwrap(); + let publish_splits_request = PublishSplitsRequest { + index_uid: Some(index_uid.clone()), + staged_split_ids: vec!["published-split".to_string()], + replaced_split_ids: Vec::new(), + index_checkpoint_delta_json_opt: None, + publish_token_opt: None, + }; + metastore + .publish_splits(publish_splits_request) + .await + .unwrap(); + + let known_split_states = index_service + .list_metastore_split_states(index_uid) + .await + .unwrap(); + let expected_split_states: HashMap = + [("published-split".into(), SplitState::Published)] + .into_iter() + .collect(); + assert_eq!(known_split_states, expected_split_states); + } + + #[tokio::test] + async fn test_publish_recovered_splits_marks_replaced_splits_for_deletion() { + let metastore = metastore_for_test(); + let storage_resolver = StorageResolver::for_test(); + let mut index_service = IndexService::new(metastore.clone(), storage_resolver); + let index_config = IndexConfig::for_test("test-index", "ram://indexes/test-index"); + let index_uid = index_service + .create_index(index_config, false) + .await + .unwrap() + .index_uid; + + let stage_splits_request = StageSplitsRequest::try_from_splits_metadata( + index_uid.clone(), + [SplitMetadata::for_test("merged-split".into())], + ) + .unwrap(); + metastore.stage_splits(stage_splits_request).await.unwrap(); + let publish_splits_request = PublishSplitsRequest { + index_uid: Some(index_uid.clone()), + staged_split_ids: vec!["merged-split".to_string()], + replaced_split_ids: Vec::new(), + index_checkpoint_delta_json_opt: None, + publish_token_opt: None, + }; + metastore + .publish_splits(publish_splits_request) + .await + .unwrap(); + + let split_to_publish = SplitToPublish { + split_metadata: SplitMetadata::for_test("merge-result-split".into()), + replaced_split_ids: vec!["merged-split".into()], + }; + index_service + .publish_recovered_splits(&index_uid, &[split_to_publish]) + .await + .unwrap(); + + let split_states = index_service + .list_metastore_split_states(index_uid) + .await + .unwrap(); + let expected_split_states: HashMap = [ + ("merge-result-split".into(), SplitState::Published), + ("merged-split".into(), SplitState::MarkedForDeletion), + ] + .into_iter() + .collect(); + assert_eq!(split_states, expected_split_states); + } + + /// Builds a split recovered from the storage of `recovered-index`, stamped with the index + /// incarnation that preceded the one the metastore holds today. + fn recovered_split(split_id: &str, parent_split_ids: &[&str]) -> RecoveredSplit { + let mut split_metadata = SplitMetadata::for_test(split_id.into()); + split_metadata.index_uid = IndexUid::for_test("recovered-index", 1); + split_metadata.num_docs = 10; + RecoveredSplit { + split_metadata, + parent_split_ids: parent_split_ids.iter().map(|id| (*id).into()).collect(), + } + } + + fn known_split_states( + split_states: impl IntoIterator, + ) -> HashMap { + split_states + .into_iter() + .map(|(split_id, split_state)| (split_id.into(), split_state)) + .collect() + } + + fn publishable_split_ids(plan: &SplitRecoveryPlan) -> Vec<&str> { + plan.splits_to_publish + .iter() + .map(|split_to_publish| split_to_publish.split_metadata.split_id().as_str()) + .collect() + } + + #[test] + fn test_plan_split_recovery_restamps_index_uid() { + let index_uid = IndexUid::for_test("recovered-index", 2); + let plan = plan_split_recovery( + &index_uid, + vec![recovered_split("split-1", &[])], + &HashMap::new(), + ); + assert_eq!(plan.splits_to_publish.len(), 1); + assert_eq!( + plan.splits_to_publish[0].split_metadata.index_uid, + index_uid + ); + } + + #[test] + fn test_plan_split_recovery_skips_splits_of_another_index() { + let index_uid = IndexUid::for_test("recovered-index", 2); + let mut foreign_split = recovered_split("split-1", &[]); + foreign_split.split_metadata.index_uid = IndexUid::for_test("another-index", 1); + let plan = plan_split_recovery(&index_uid, vec![foreign_split], &HashMap::new()); + assert!(plan.splits_to_publish.is_empty()); + } + + #[test] + fn test_plan_split_recovery_skips_splits_superseded_by_a_merge_chain() { + // `split-3` was merged out of `split-2`, itself merged out of `split-1`: all three files + // can still be in the storage, but only the last one may be published. + let index_uid = IndexUid::for_test("recovered-index", 2); + let plan = plan_split_recovery( + &index_uid, + vec![ + recovered_split("split-1", &[]), + recovered_split("split-2", &["split-1"]), + recovered_split("split-3", &["split-2"]), + ], + &HashMap::new(), + ); + assert_eq!(publishable_split_ids(&plan), ["split-3"]); + assert_eq!(plan.merged_split_ids.len(), 2); + assert!(plan.already_known_split_ids.is_empty()); + assert!(plan.splits_to_publish[0].replaced_split_ids.is_empty()); + } + + #[test] + fn test_plan_split_recovery_skips_splits_already_in_the_metastore() { + let index_uid = IndexUid::for_test("recovered-index", 2); + let known_split_states = known_split_states([("split-1", SplitState::Published)]); + let plan = plan_split_recovery( + &index_uid, + vec![ + recovered_split("split-1", &[]), + recovered_split("split-2", &[]), + ], + &known_split_states, + ); + assert_eq!(publishable_split_ids(&plan), ["split-2"]); + assert_eq!(plan.already_known_split_ids.len(), 1); + } + + #[test] + fn test_plan_split_recovery_replaces_published_ancestors() { + // The metastore was restored from a backup taken before the merges: publishing + // `split-4` must retire the splits it was merged out of, transitively. + let index_uid = IndexUid::for_test("recovered-index", 2); + let known_split_states = known_split_states([ + ("split-1", SplitState::Published), + ("split-2", SplitState::Published), + ("split-3", SplitState::Published), + ]); + let plan = plan_split_recovery( + &index_uid, + vec![ + recovered_split("split-3", &["split-1", "split-2"]), + recovered_split("split-4", &["split-3"]), + ], + &known_split_states, + ); + assert_eq!(publishable_split_ids(&plan), ["split-4"]); + let mut replaced_split_ids: Vec<&str> = plan.splits_to_publish[0] + .replaced_split_ids + .iter() + .map(SplitId::as_str) + .collect(); + replaced_split_ids.sort(); + assert_eq!(replaced_split_ids, ["split-1", "split-2", "split-3"]); + } + + #[test] + fn test_plan_split_recovery_does_not_replace_unpublished_ancestors() { + // Replacing a split that the metastore does not hold as published is rejected, and an + // ancestor already marked for deletion is on its way out anyway. + let index_uid = IndexUid::for_test("recovered-index", 2); + let known_split_states = known_split_states([("split-1", SplitState::MarkedForDeletion)]); + let plan = plan_split_recovery( + &index_uid, + vec![recovered_split("split-3", &["split-1", "split-2"])], + &known_split_states, + ); + assert_eq!(publishable_split_ids(&plan), ["split-3"]); + assert!(plan.splits_to_publish[0].replaced_split_ids.is_empty()); + } + + #[test] + fn test_plan_split_recovery_replaces_an_ancestor_at_most_once() { + // `split-1` was merged twice: the first merge result was never published, so both + // results sit in the storage. The metastore only accepts it as replaced once. + let index_uid = IndexUid::for_test("recovered-index", 2); + let known_split_states = known_split_states([("split-1", SplitState::Published)]); + let plan = plan_split_recovery( + &index_uid, + vec![ + recovered_split("split-2", &["split-1"]), + recovered_split("split-3", &["split-1"]), + ], + &known_split_states, + ); + assert_eq!(publishable_split_ids(&plan), ["split-2", "split-3"]); + let replaced_split_ids: Vec<&str> = plan + .splits_to_publish + .iter() + .flat_map(|split_to_publish| split_to_publish.replaced_split_ids.iter()) + .map(SplitId::as_str) + .collect(); + assert_eq!(replaced_split_ids, ["split-1"]); + } } diff --git a/quickwit/quickwit-index-management/src/lib.rs b/quickwit/quickwit-index-management/src/lib.rs index b8ab6ea2850..a95c7b1071b 100644 --- a/quickwit/quickwit-index-management/src/lib.rs +++ b/quickwit/quickwit-index-management/src/lib.rs @@ -16,4 +16,7 @@ mod garbage_collection; mod index; pub use garbage_collection::{GcMetrics, SplitRemovalInfo, run_garbage_collect}; -pub use index::{IndexService, IndexServiceError, clear_cache_directory, validate_storage_uri}; +pub use index::{ + IndexService, IndexServiceError, SplitRecoveryInfo, SplitToPublish, UnreadableSplit, + clear_cache_directory, validate_storage_uri, +};