דחיפת נתונים בכתיבת אובייקטים ב-Cloud Storage

הלקוח של Cloud Storage בספריות הלקוח ב-Cloud ל-Rust כותב אובייקטים על ידי משיכת קטעי מטען ייעודי (payload) מסוג שמטמיע את המאפיין StreamingSource. חלק מהאפליקציות יוצרות נתונים במשימה או בשרשור נפרדים, וצריכות לדחוף חלקי מטען ייעודי (payload) להעלאה.

במאמר הזה מוסבר איך לעטוף מקלט ערוץ אסינכרוני בהטמעה של StreamingSource כדי שתוכלו לדחוף נתונים לכתיבת אובייקט ב-Cloud Storage. אם אתם משתמשים חדשים בספריית הלקוח, קודם קראו את המאמר שימוש ב-Cloud Storage עם Rust.

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

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

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

יחסי תלות

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

cargo add google-cloud-storage anyhow bytes
cargo add tokio --features rt,sync

המרת מקלט של תור למקור סטרימינג

כדי להפריד בין המשימה שמפיקה את הנתונים לבין המשימה שמעלה את המטען הייעודי (payload), חברו אותן באמצעות תור אסינכרוני. במסמך הזה אנחנו משתמשים בערוץ mpsc של Tokio, אבל אתם יכולים להשתמש בכל ערוץ או תור שמשתלבים עם זמן הריצה האסינכרוני שלכם.

כדי להתאים את מקלט הערוץ להטמעה של StreamingSource, מבצעים את השלבים הבאים:

1. מגדירים מבנה שעוטף את הערוץ Receiver:

   use tokio::sync::mpsc::{self, Receiver};
   #[derive(Debug)]
   struct QueueSource(Receiver<bytes::Bytes>);

2. מטמיעים את המאפיין StreamingSource עבור ה-struct של ה-wrapper, כך שכל קריאה ל-next תמתין לחלק הבא מהמקבל:

   use google_cloud_storage::streaming_source::StreamingSource;
   impl StreamingSource for QueueSource {
       type Error = std::convert::Infallible;
       async fn next(&mut self) -> Option<Result<bytes::Bytes, Self::Error>> {
           self.0.recv().await.map(Ok)
       }
   }

העברת נתונים בסטרימינג מהתור אל Cloud Storage

אחרי שמטמיעים את המאפיין StreamingSource, אפשר להפעיל את העלאת האובייקט במשימה ברקע ולשלוח מקטעים דרך הערוץ ממשימת ה-producer. מכיוון שמקלט הערוץ הוא מקור נתונים במעבר יחיד שלא מטמיע את Seek, צריך להתחיל את ההעלאה באמצעות send_buffered במקום send_unbuffered.

בתוך פונקציית async שמקבלת את bucket_name בפורמט projects/_/buckets/BUCKET_ID ו-object_name, מבצעים את הפעולות הבאות:

1. מאתחלים לקוח Storage:

   use google_cloud_storage::client::Storage;
   let client = Storage::builder().build().await?;

2. יוצרים ערוץ mpsc מוגבל כדי לקבל את נקודות האחיזה sender ו-receiver:

   let (sender, receiver) = mpsc::channel::<bytes::Bytes>(32);

3. מגדירים את העלאת האובייקט עם QueueSource(receiver) כמקור המטען הייעודי (payload). אל תחכו עדיין לעתיד של send_buffered:

   let upload = client
       .write_object(bucket_name, object_name, QueueSource(receiver))
       .send_buffered();

4. מפעילים משימה ברקע כדי להעלות את הנתונים בזמן שהיצרן יוצר אותם:

   let task = tokio::spawn(upload);

5. במשימת ה-producer, דוחפים מקטעי נתונים לכינוי sender:

   for _ in 0..1000 {
       let line = "I will not write funny examples in class\n";
       sender
           .send(bytes::Bytes::from_static(line.as_bytes()))
           .await?;
   }

6. אחרי ששולחים את כל הנתונים, סוגרים את הערוץ באמצעות ה-handle של sender ומסמנים את סוף הסטרימינג של האובייקט:

   drop(sender);

7. ממתינים לטיפול במשימה ברקע כדי לוודא שהמשימה ובקשת ההעלאה הצליחו, ומאחזרים את המטא-נתונים של Object שהועלו:

   let object = task.await??;
   println!("object successfully uploaded {object:?}");

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

הסרת המשאבים

כדי להימנע מחיובים בחשבון Google Cloud על המשאבים שבהם השתמשתם במסמך הזה, מחקו את אובייקטי הבדיקה שהעליתם כשתסיימו.

השלמת הקוד

בדוגמה הבאה משולבים כל השלבים הקודמים לפונקציה מלאה:

use google_cloud_storage::streaming_source::StreamingSource;
use tokio::sync::mpsc::{self, Receiver};
#[derive(Debug)]
struct QueueSource(Receiver<bytes::Bytes>);
impl StreamingSource for QueueSource {
    type Error = std::convert::Infallible;
    async fn next(&mut self) -> Option<Result<bytes::Bytes, Self::Error>> {
        self.0.recv().await.map(Ok)
    }
}

pub async fn queue(bucket_name: &str, object_name: &str) -> anyhow::Result<()> {
    use google_cloud_storage::client::Storage;
    let client = Storage::builder().build().await?;

    let (sender, receiver) = mpsc::channel::<bytes::Bytes>(32);
    let upload = client
        .write_object(bucket_name, object_name, QueueSource(receiver))
        .send_buffered();
    let task = tokio::spawn(upload);

    for _ in 0..1000 {
        let line = "I will not write funny examples in class\n";
        sender
            .send(bytes::Bytes::from_static(line.as_bytes()))
            .await?;
    }
    drop(sender);
    let object = task.await??;
    println!("object successfully uploaded {object:?}");

    Ok(())
}