כתיבה מחדש של אובייקטים ב-Cloud Storage

כתיבה מחדש של אובייקט ב-Cloud Storage מעתיקה או משנה אובייקט קיים, למשל מעבירה אותו בין מיקומים, מחליפה את מפתח ההצפנה שלו או משנה את סוג האחסון שלו. בהתאם לגודל האובייקט ולפרמטרים של הכתיבה מחדש, יכול להיות ש-Cloud Storage ישלים את הפעולה בקריאה אחת או יחזיר התקדמות ביניים עם rewrite_token. כשקריאה מחזירה את הערך rewrite_token, מעבירים את האסימון הזה לבקשות הבאות עד שהשכתוב מסתיים.

במאמר הזה מוסבר איך להגדיר בקשת שכתוב ולהפעיל את לולאת השכתוב באמצעות פונקציית העזר rewrite_until_done או לולאה ידנית בספריות הלקוח של Cloud ל-Rust. אם אתם משתמשים חדשים בספריית הלקוח, קודם כדאי לקרוא את המאמר שימוש ב-Cloud Storage עם Rust.

לפני שמתחילים

לפני שמריצים את הקוד במסמך הזה, צריך להשלים את הדרישות המוקדמות הבאות:

• יוצרים או בוחרים פרויקט ב-Google Cloud שהחיוב מופעל בו.
• יוצרים קטגוריה של Cloud Storage לבדיקה ייעודית בפרויקט.
• מגדירים את Application Default Credentials בסביבה שלכם.
• מקצים לחשבון המשתמש תפקיד בניהול זהויות והרשאות גישה (IAM) עם הרשאה לקרוא, לכתוב ולמחוק אובייקטים בקטגוריה, כמו אדמין של אחסון (roles/storage.admin).

יחסי תלות

מוסיפים את ספריית הלקוח של Cloud Storage ואת ה-crates התומכים לקובץ Cargo.toml:

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

הכנת הלקוח ובקשת השכתוב

גם העוזר האוטומטי וגם הלולאה הידנית מתחילים מלקוח StorageControl ומבונה בקשות RewriteObject שהוגדרו. בתוך היקף שמקבל את bucket_name בפורמט projects/_/buckets/BUCKET_ID, מבצעים את הפעולות הבאות:

1. מאתחלים לקוח StorageControl עם מדיניות ניסיון חוזר. מומלץ להמתין לפחות 30 שניות בין בקשה לבקשה. בדוגמה הזו משתמשים ב-RetryableErrors עם מגבלה של 5 ניסיונות וללא מגבלת זמן כוללת לכל פעולה:

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

2. יוצרים את ה-request builder של 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");

3. אופציונלי: הגבלת מספר הבייטים המקסימלי לשכתוב בכל קריאה, כדי שבקשות ספציפיות יחזירו התקדמות מהר יותר במקום לדרוש זמן קצוב ארוך יותר לבקשה. בדוגמה הזו מוגדרת מגבלה קטנה של 1 MiB כדי לאלץ כמה חזרות. בסביבת הייצור צריך להשתמש בערך גדול יותר:

   builder = builder.set_max_bytes_rewritten_per_call(1024 * 1024);

4. אופציונלי: מציינים את המטא-נתונים של אובייקט היעד, כמו שינוי סוג האחסון ל-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:?}");

הפעלה ידנית של לולאת השכתוב

אם אתם צריכים לתעד ספירות בייטים של ביניים, לבדוק את ההתקדמות או להמשיך שכתוב מתהליך אחר, הפעילו את לולאת השכתוב באופן ידני:

1. מגדירים פונקציית עזר ששולחת בקשת RewriteObject אחת, מתעדת את ההתקדמות ומחזירה את המטא-נתונים של Object שהושלמו כש-resp.done הוא true, או את ה-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. מפעילים את פונקציית העזר בלולאה, משכפלים את הכלי ליצירת בקשות בכל איטרציה ומעבירים את האסימון שהוחזר אל set_rewrite_token בכל פעם ש-make_one_request מחזירה את הערך 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:?}");

הערה: טוקנים של כתיבה מחדש נשארים בתוקף למשך עד שבוע, כך שאפשר לאחסן rewrite_token ולהמשיך כתיבה מחדש שהופסקה מתהליך אחר עם פרמטרים תואמים של בקשה.

השלמת הקוד

בדוגמה הבאה, אובייקט מקור בגודל 3 MiB מועלה ומשוכתב לסוג האחסון Nearline Storage במרווחים של 1 MiB באמצעות הלולאה הידנית. אחרי שהכתיבה מחדש מסתיימת, הדוגמה מוחקת את אובייקטי הבדיקה ואת קטגוריית הבדיקה הריקה:

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 .