risingwave_meta/backup_restore/
restore.rs

1// Copyright 2025 RisingWave Labs
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::collections::HashSet;
16use std::sync::Arc;
17
18use anyhow::anyhow;
19use futures::TryStreamExt;
20use risingwave_backup::MetaSnapshotId;
21use risingwave_backup::error::{BackupError, BackupResult};
22use risingwave_backup::meta_snapshot::Metadata;
23use risingwave_backup::storage::{MetaSnapshotStorage, MetaSnapshotStorageRef};
24use risingwave_common::config::{MetaBackend, ObjectStoreConfig};
25use risingwave_hummock_sdk::version::HummockVersion;
26use risingwave_hummock_sdk::{
27    HummockRawObjectId, try_get_object_id_from_path, version_checkpoint_path,
28};
29use risingwave_object_store::object::build_remote_object_store;
30use risingwave_object_store::object::object_metrics::ObjectStoreMetrics;
31use risingwave_pb::hummock::PbHummockVersionCheckpoint;
32use thiserror_ext::AsReport;
33
34use crate::backup_restore::restore_impl::v2::{LoaderV2, WriterModelV2ToMetaStoreV2};
35use crate::backup_restore::restore_impl::{Loader, Writer};
36use crate::backup_restore::utils::{get_backup_store, get_meta_store};
37use crate::controller::SqlMetaStore;
38
39/// Command-line arguments for restore.
40#[derive(clap::Args, Debug, Clone)]
41pub struct RestoreOpts {
42    /// Id of snapshot used to restore. Available snapshots can be found in
43    /// <`storage_directory>/manifest.json`.
44    #[clap(long)]
45    pub meta_snapshot_id: u64,
46    /// Type of meta store to restore.
47    #[clap(long, value_enum, default_value_t = MetaBackend::Mem)]
48    pub meta_store_type: MetaBackend,
49    #[clap(long, default_value_t = String::from(""))]
50    pub sql_endpoint: String,
51    /// Username of sql backend, required when meta backend set to MySQL or PostgreSQL.
52    #[clap(long, default_value = "")]
53    pub sql_username: String,
54    /// Password of sql backend, required when meta backend set to MySQL or PostgreSQL.
55    #[clap(long, default_value = "")]
56    pub sql_password: String,
57    /// Database of sql backend, required when meta backend set to MySQL or PostgreSQL.
58    #[clap(long, default_value = "")]
59    pub sql_database: String,
60    /// Url of storage to fetch meta snapshot from.
61    #[clap(long)]
62    pub backup_storage_url: String,
63    /// Directory of storage to fetch meta snapshot from.
64    #[clap(long, default_value_t = String::from("backup"))]
65    pub backup_storage_directory: String,
66    /// Url of storage to restore hummock version to.
67    #[clap(long)]
68    pub hummock_storage_url: String,
69    /// Directory of storage to restore hummock version to.
70    #[clap(long, default_value_t = String::from("hummock_001"))]
71    pub hummock_storage_directory: String,
72    /// Print the target snapshot, but won't restore to meta store.
73    #[clap(long, default_value_t = false)]
74    pub dry_run: bool,
75    /// The read timeout for object store
76    #[clap(long, default_value_t = 600000)]
77    pub read_attempt_timeout_ms: u64,
78    /// The maximum number of read retry attempts for the object store.
79    #[clap(long, default_value_t = 3)]
80    pub read_retry_attempts: u64,
81    #[clap(long, default_value_t = false)]
82    /// When enabled, some system parameters of in the restored meta store will be overwritten.
83    /// Specifically, system parameters `state_store`, `data_directory`, `backup_storage_url` and `backup_storage_directory` will be overwritten
84    /// with the specified opts `hummock_storage_url`, `hummock_storage_directory`, `overwrite_backup_storage_url` and `overwrite_backup_storage_directory`.
85    pub overwrite_hummock_storage_endpoint: bool,
86    #[clap(long, required = false)]
87    pub overwrite_backup_storage_url: Option<String>,
88    #[clap(long, required = false)]
89    pub overwrite_backup_storage_directory: Option<String>,
90    /// Verify that all referenced objects exist in object store.
91    #[clap(long, default_value_t = false)]
92    pub validate_integrity: bool,
93}
94
95async fn restore_hummock_version(
96    hummock_storage_url: &str,
97    hummock_storage_directory: &str,
98    hummock_version: &HummockVersion,
99) -> BackupResult<()> {
100    let object_store = Arc::new(
101        build_remote_object_store(
102            hummock_storage_url,
103            Arc::new(ObjectStoreMetrics::unused()),
104            "Version Checkpoint",
105            Arc::new(ObjectStoreConfig::default()),
106        )
107        .await,
108    );
109    let checkpoint_path = version_checkpoint_path(hummock_storage_directory);
110    let checkpoint = PbHummockVersionCheckpoint {
111        version: Some(hummock_version.into()),
112        // Ignore stale objects. Full GC will clear them.
113        stale_objects: Default::default(),
114    };
115    use prost::Message;
116    let buf = checkpoint.encode_to_vec();
117    object_store
118        .upload(&checkpoint_path, buf.into())
119        .await
120        .map_err(|e| BackupError::StateStorage(e.into()))?;
121    Ok(())
122}
123
124/// Restores a meta store.
125/// Uses `meta_store` and `backup_store` if provided.
126/// Otherwise creates them based on `opts`.
127async fn restore_impl(
128    opts: RestoreOpts,
129    meta_store: Option<SqlMetaStore>,
130    backup_store: Option<MetaSnapshotStorageRef>,
131) -> BackupResult<()> {
132    if cfg!(not(test)) {
133        assert!(meta_store.is_none());
134        assert!(backup_store.is_none());
135    }
136    let meta_store = match meta_store {
137        None => get_meta_store(opts.clone()).await?,
138        Some(m) => m,
139    };
140    let backup_store = match backup_store {
141        None => get_backup_store(opts.clone()).await?,
142        Some(b) => b,
143    };
144    let target_id = opts.meta_snapshot_id;
145    let snapshot_list = &backup_store.manifest().snapshot_metadata;
146    let snapshot = match snapshot_list.iter().find(|m| m.id == target_id) {
147        None => {
148            return Err(BackupError::Other(anyhow::anyhow!(
149                "snapshot id {} not found",
150                target_id
151            )));
152        }
153        Some(s) => s,
154    };
155
156    if opts.validate_integrity {
157        tracing::info!("Start integrity validation.");
158        validate_integrity(
159            snapshot.objects.clone(),
160            &opts.hummock_storage_url,
161            &opts.hummock_storage_directory,
162        )
163        .await
164        .inspect_err(|_| tracing::error!("Fail integrity validation."))?;
165        tracing::info!("Succeed integrity validation.");
166    }
167
168    let format_version = snapshot.format_version;
169    if format_version < 2 {
170        unimplemented!("not supported: write model V1 to meta store V2");
171    } else {
172        dispatch(
173            target_id,
174            &opts,
175            LoaderV2::new(backup_store),
176            WriterModelV2ToMetaStoreV2::new(meta_store.to_owned()),
177        )
178        .await?;
179    }
180
181    Ok(())
182}
183
184async fn dispatch<L: Loader<S>, W: Writer<S>, S: Metadata>(
185    target_id: MetaSnapshotId,
186    opts: &RestoreOpts,
187    loader: L,
188    writer: W,
189) -> BackupResult<()> {
190    // Validate parameters.
191    if opts.overwrite_hummock_storage_endpoint
192        && (opts.overwrite_backup_storage_url.is_none()
193            || opts.overwrite_backup_storage_directory.is_none())
194    {
195        return Err(BackupError::Other(anyhow::anyhow!("overwrite_hummock_storage_endpoint, overwrite_backup_storage_url, overwrite_backup_storage_directory must be set simultaneously".to_owned())));
196    }
197
198    // Restore meta store.
199    let target_snapshot = loader.load(target_id).await?;
200    if opts.dry_run {
201        tracing::info!("Complete dry run.");
202        return Ok(());
203    }
204    let hummock_version = target_snapshot.metadata.hummock_version_ref().clone();
205    writer.write(target_snapshot).await?;
206    if opts.overwrite_hummock_storage_endpoint {
207        writer
208            .overwrite(
209                &format!("hummock+{}", opts.hummock_storage_url),
210                &opts.hummock_storage_directory,
211                opts.overwrite_backup_storage_url.as_ref().unwrap(),
212                opts.overwrite_backup_storage_directory.as_ref().unwrap(),
213            )
214            .await?;
215    }
216
217    // Restore object store.
218    restore_hummock_version(
219        &opts.hummock_storage_url,
220        &opts.hummock_storage_directory,
221        &hummock_version,
222    )
223    .await?;
224    Ok(())
225}
226
227pub async fn restore(opts: RestoreOpts) -> BackupResult<()> {
228    tracing::info!("restore with opts: {:#?}", opts);
229    let result = restore_impl(opts, None, None).await;
230    match &result {
231        Ok(_) => {
232            tracing::info!("command succeeded");
233        }
234        Err(e) => {
235            tracing::warn!(error = %e.as_report(), "command failed");
236        }
237    }
238    result
239}
240
241async fn validate_integrity(
242    mut object_ids: HashSet<HummockRawObjectId>,
243    hummock_storage_url: &str,
244    hummock_storage_directory: &str,
245) -> BackupResult<()> {
246    tracing::info!("expect {} objects", object_ids.len());
247    let object_store = Arc::new(
248        build_remote_object_store(
249            hummock_storage_url,
250            Arc::new(ObjectStoreMetrics::unused()),
251            "Version Checkpoint",
252            Arc::new(ObjectStoreConfig::default()),
253        )
254        .await,
255    );
256    let mut iter = object_store
257        .list(hummock_storage_directory, None, None)
258        .await?;
259    while let Some(obj) = iter.try_next().await? {
260        let Some(obj_id) = try_get_object_id_from_path(&obj.key) else {
261            continue;
262        };
263        if object_ids.remove(&obj_id.as_raw()) && object_ids.is_empty() {
264            break;
265        }
266    }
267    if object_ids.is_empty() {
268        return Ok(());
269    }
270    Err(BackupError::Other(anyhow!(
271        "referenced objects not found in object store: {:?}",
272        object_ids
273    )))
274}
risingwave_meta/backup_restore/restore.rs

risingwave_meta/backup_restore/
restore.rs
