Reescribe objetos de Cloud Storage

Reescribir un objeto en Cloud Storage copia o transforma un objeto existente, como moverlo entre ubicaciones, rotar su clave de encriptación o cambiar su clase de almacenamiento. Según el tamaño del objeto y los parámetros de reescritura, Cloud Storage puede completar la operación en una sola llamada o devolver el progreso intermedio con un rewrite_token. Cuando una llamada devuelva un rewrite_token, pasa ese token a las solicitudes posteriores hasta que finalice la reescritura.

En este documento, se muestra cómo configurar una solicitud de reescritura y controlar el bucle de reescritura con el asistente rewrite_until_done o un bucle manual en las bibliotecas cliente de Cloud para Rust. Si es la primera vez que usas la biblioteca cliente, consulta Usa Cloud Storage con Rust primero.

Antes de comenzar

Antes de ejecutar el código de este documento, completa los siguientes requisitos previos:

• Crea o selecciona un proyecto deGoogle Cloud con la facturación habilitada.
• Crea un bucket de Cloud Storage de prueba dedicado en tu proyecto.
• Configura las credenciales predeterminadas de la aplicación para tu entorno.
• Otorga a tu principal un rol de Identity and Access Management (IAM) con permiso para leer, escribir y borrar objetos en el bucket, como Administrador de almacenamiento (roles/storage.admin).

Dependencias

Agrega la biblioteca cliente de Cloud Storage y los crates de compatibilidad a tu archivo Cargo.toml:

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

Prepara el cliente y la solicitud de reescritura

Tanto el ayudante automático como el bucle manual comienzan desde un cliente StorageControl y un compilador de solicitudes RewriteObject configurados. Dentro de un permiso que acepte bucket_name en el formato projects/_/buckets/BUCKET_ID, completa los siguientes pasos:

1. Inicializa un cliente de StorageControl con una política de reintentos. Recomendamos permitir al menos 30 segundos por solicitud. En este ejemplo, se usa RetryableErrors con un límite de 5 intentos y sin un límite de tiempo general por operación:

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

2. Crea el compilador de solicitudes rewrite_object, marca la solicitud como idempotente para que el cliente pueda reintentar los errores transitorios y establece los nombres de bucket y objetos de origen y destino:

   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");

3. Opcional: Limita la cantidad máxima de bytes que se reescriben por llamada para que las solicitudes individuales devuelvan el progreso más rápido en lugar de requerir un tiempo de espera de solicitud más largo. En este ejemplo, se establece un límite pequeño de 1 MiB para forzar varias iteraciones. Usa un valor mayor en producción:

   builder = builder.set_max_bytes_rewritten_per_call(1024 * 1024);

4. Opcional: Especifica los metadatos del objeto de destino, como cambiar la clase de almacenamiento a Nearline (NEARLINE):

   builder = builder.set_destination(Object::new().set_storage_class("NEARLINE"));

   Nota: Las clases de almacenamiento Nearline, Coldline y Archive aplican una duración mínima de almacenamiento. Es posible que se te cobren cargos por eliminación anticipada si borras o reemplazas objetos antes de que finalice ese período.

Ejecuta el bucle de reescritura automáticamente

En la mayoría de los casos de uso, incluye el atributo RewriteObjectExt en el alcance y llama a rewrite_until_done en el compilador de solicitudes. Este ayudante administra el rewrite_token en las iteraciones y se resuelve en los metadatos Object de destino después de que se completa la reescritura:

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

Ejecuta el bucle de reescritura de forma manual

Si necesitas registrar recuentos de bytes intermedios, el progreso de los puntos de control o reanudar una reescritura desde otro proceso, controla el bucle de reescritura de forma manual:

1. Define una función auxiliar que envíe una sola solicitud RewriteObject, registre el progreso y devuelva los metadatos de Object completados cuando resp.done sea true o el siguiente 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))
   }

2. Llama al ayudante en un bucle, clona el compilador de solicitudes en cada iteración y pasa el token devuelto a set_rewrite_token cada vez que make_one_request devuelva RewriteProgress::Incomplete:

   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:?}");

Nota: Los tokens de reescritura siguen siendo válidos durante un máximo de una semana, por lo que puedes almacenar un rewrite_token y reanudar una reescritura interrumpida desde otro proceso con parámetros de solicitud coincidentes.

Completar código

En el siguiente ejemplo, se sube un objeto de origen de 3 MiB y se vuelve a escribir en la clase de almacenamiento Nearline Storage en incrementos de 1 MiB con el bucle manual. Una vez que finaliza la reescritura, en el ejemplo se borran los objetos de prueba y el bucket de prueba vacío:

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;
}

El resultado es similar al siguiente, que muestra el progreso en tres iteraciones y los metadatos del objeto de destino:

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", ... }

Realiza una limpieza

Si la muestra se cierra antes de tiempo o ejecutas fragmentos individuales sin el asistente de limpieza, borra los objetos de prueba restantes y borra el bucket de prueba cuando termines para que no se generen cargos continuos en tu cuenta de Google Cloud .