Google 会使用 AI 技术将内容翻译成您偏好的语言。AI 翻译可能包含错误。

重写 Cloud Storage 对象

在 Cloud Storage 中重写对象会复制或转换现有对象，例如跨位置移动对象、轮替其加密密钥或更改其存储类别。根据对象大小和重写参数，Cloud Storage 可能会在一次调用中完成操作，也可能会返回中间进度和 rewrite_token。当调用返回 rewrite_token 时，请将该令牌传递给后续请求，直到重写完成。

本文档介绍了如何使用适用于 Rust 的 Cloud 客户端库中的 rewrite_until_done 辅助程序或手动循环来配置重写请求并驱动重写循环。如果您刚开始接触客户端库，请先参阅将 Cloud Storage 与 Rust 搭配使用。

准备工作

在运行本文档中的代码之前，请先完成以下前提条件：

创建或选择一个已启用结算功能的 Google Cloud 项目。
在项目中创建专用的测试 Cloud Storage 存储桶。
为您的环境配置应用默认凭证。
向您的主账号授予具有在存储桶中读取、写入和删除对象权限的 Identity and Access Management (IAM) 角色，例如 Storage Admin (roles/storage.admin)。

依赖项

将 Cloud Storage 客户端库和支持 crate 添加到 Cargo.toml 文件中：

cargo add google-cloud-storage google-cloud-gax anyhow bytes
cargo add tokio --features macros,rt-multi-thread

准备客户端和重写请求

自动辅助程序和手动循环都从配置的 StorageControl 客户端和 RewriteObject 请求构建器开始。在接受格式为 projects/_/buckets/BUCKET_ID 的 bucket_name 的范围内，完成以下步骤：

使用重试政策初始化 StorageControl 客户端。我们建议每个请求至少留出 30 秒。此示例使用 RetryableErrors，尝试次数上限为 5 次，且没有每次操作的总时间限制：
```
let control = StorageControl::builder()
    .with_retry_policy(RetryableErrors.with_attempt_limit(5))
    .build()
    .await?;
```

创建 rewrite_object 请求构建器，将请求标记为幂等，以便客户端可以重试暂时性错误，并设置来源和目标存储桶及对象名称：

let mut builder = control
    .rewrite_object()
    .with_idempotency(true) // retry transient errors
    .set_source_bucket(bucket_name)
    .set_source_object(&source_object.name)
    .set_destination_bucket(bucket_name)
    .set_destination_name("rewrite-object-clone");

可选：限制每次调用重写的最大字节数，以便单个请求更快地返回进度，而不是需要更长的请求超时。此示例设置了 1 MiB 的小限制，以强制执行多次迭代；在生产环境中使用更大的值：
```
builder = builder.set_max_bytes_rewritten_per_call(1024 * 1024);
```
可选：指定目标对象元数据，例如将存储类别更改为 Nearline (NEARLINE)：
```
builder = builder.set_destination(Object::new().set_storage_class("NEARLINE"));
```
注意：Nearline、Coldline 和 Archive 存储类别强制执行最短存储期限。如果您在该期限结束之前删除或覆盖对象，则可能会产生提前删除费用。

自动运行重写循环

对于大多数用例，请将 RewriteObjectExt 特征引入范围，并在请求构建器上调用 rewrite_until_done。此辅助程序会跨迭代管理 rewrite_token，并在重写完成后解析为目标 Object 元数据：

use google_cloud_storage::builder_ext::RewriteObjectExt;
let dest_object = builder.rewrite_until_done().await?;
println!("dest_object={dest_object:?}");

手动运行重写循环

如果您需要记录中间字节数、检查点进度或从其他进程恢复重写，请手动驱动重写循环：

定义一个辅助函数，用于发送单个 RewriteObject 请求、记录进度，并在 resp.done 为 true 时返回已完成的 Object 元数据，或返回下一个 rewrite_token：

enum RewriteProgress {
    // This holds the rewrite token
    Incomplete(String),
    Done(Box<Object>),
}

async fn make_one_request(builder: RewriteObject) -> Result<RewriteProgress> {
    let resp = builder.send().await?;
    if resp.done {
        println!(
            "DONE:     total_bytes_rewritten={}; object_size={}",
            resp.total_bytes_rewritten, resp.object_size
        );
        return Ok(RewriteProgress::Done(Box::new(
            resp.resource
                .expect("A `done` response must have an object."),
        )));
    }
    println!(
        "PROGRESS: total_bytes_rewritten={}; object_size={}",
        resp.total_bytes_rewritten, resp.object_size
    );
    Ok(RewriteProgress::Incomplete(resp.rewrite_token))
}

在循环中调用辅助函数，在每次迭代时克隆请求构建器，并在 make_one_request 返回 RewriteProgress::Incomplete 时将返回的令牌传递给 set_rewrite_token：

let dest_object = loop {
    let progress = make_one_request(builder.clone()).await?;
    match progress {
        RewriteProgress::Incomplete(rewrite_token) => {
            builder = builder.set_rewrite_token(rewrite_token);
        }
        RewriteProgress::Done(object) => break object,
    };
};
println!("dest_object={dest_object:?}");

完整代码

以下示例上传一个 3 MiB 的源对象，并使用手动循环以 1 MiB 的增量将其重写为 Nearline Storage 存储类别。重写完成后，该示例会删除测试对象和空测试存储桶：

use gcs::Result;
use gcs::builder::storage_control::RewriteObject;
use gcs::client::StorageControl;
use gcs::model::Object;
use gcs::retry_policy::RetryableErrors;
use google_cloud_gax::options::RequestOptionsBuilder;
use google_cloud_gax::retry_policy::RetryPolicyExt as _;
use google_cloud_storage as gcs;

pub async fn rewrite_object(bucket_name: &str) -> anyhow::Result<()> {
    let source_object = upload(bucket_name).await?;

    let control = StorageControl::builder()
        .with_retry_policy(RetryableErrors.with_attempt_limit(5))
        .build()
        .await?;

    let mut builder = control
        .rewrite_object()
        .with_idempotency(true) // retry transient errors
        .set_source_bucket(bucket_name)
        .set_source_object(&source_object.name)
        .set_destination_bucket(bucket_name)
        .set_destination_name("rewrite-object-clone");

    // Optionally limit the max bytes written per request.
    builder = builder.set_max_bytes_rewritten_per_call(1024 * 1024);

    // Optionally change the storage class to force GCS to copy bytes
    builder = builder.set_destination(Object::new().set_storage_class("NEARLINE"));

    let dest_object = loop {
        let progress = make_one_request(builder.clone()).await?;
        match progress {
            RewriteProgress::Incomplete(rewrite_token) => {
                builder = builder.set_rewrite_token(rewrite_token);
            }
            RewriteProgress::Done(object) => break object,
        };
    };
    println!("dest_object={dest_object:?}");

    cleanup(control, bucket_name, &source_object.name, &dest_object.name).await;
    Ok(())
}

enum RewriteProgress {
    // This holds the rewrite token
    Incomplete(String),
    Done(Box<Object>),
}

async fn make_one_request(builder: RewriteObject) -> Result<RewriteProgress> {
    let resp = builder.send().await?;
    if resp.done {
        println!(
            "DONE:     total_bytes_rewritten={}; object_size={}",
            resp.total_bytes_rewritten, resp.object_size
        );
        return Ok(RewriteProgress::Done(Box::new(
            resp.resource
                .expect("A `done` response must have an object."),
        )));
    }
    println!(
        "PROGRESS: total_bytes_rewritten={}; object_size={}",
        resp.total_bytes_rewritten, resp.object_size
    );
    Ok(RewriteProgress::Incomplete(resp.rewrite_token))
}

// Upload an object to rewrite
async fn upload(bucket_name: &str) -> anyhow::Result<Object> {
    let storage = gcs::client::Storage::builder().build().await?;
    // We need the size to exceed 1MiB to exercise the rewrite token logic.
    let payload = bytes::Bytes::from(vec![65_u8; 3 * 1024 * 1024]);
    let object = storage
        .write_object(bucket_name, "rewrite-object-source", payload)
        .send_unbuffered()
        .await?;
    Ok(object)
}

// Clean up the resources created in this sample
async fn cleanup(control: StorageControl, bucket_name: &str, o1: &str, o2: &str) {
    let _ = control
        .delete_object()
        .set_bucket(bucket_name)
        .set_object(o1)
        .send()
        .await;
    let _ = control
        .delete_object()
        .set_bucket(bucket_name)
        .set_object(o2)
        .send()
        .await;
    let _ = control.delete_bucket().set_name(bucket_name).send().await;
}

输出类似于以下内容，显示了三次迭代的进度和目标对象元数据：

PROGRESS: total_bytes_rewritten=1048576; object_size=3145728
PROGRESS: total_bytes_rewritten=2097152; object_size=3145728
DONE:     total_bytes_rewritten=3145728; object_size=3145728
dest_object=Object { name: "rewrite-object-clone", ... }

清理

如果示例提前退出，或者您在没有清理帮助程序的情况下运行单个代码段，请删除所有剩余的测试对象，并在完成后删除测试存储桶，以免您的 Google Cloud 账号产生持续费用。