vicky-worker: chunk log uploads

wobcom · Oct 4, 2023 · 51285fb · 51285fb
1 parent ba8d39c
commit 51285fb
Showing 1 changed file with 8 additions and 3 deletions.
diff --git a/vicky-worker/src/main.rs b/vicky-worker/src/main.rs
@@ -75,14 +75,14 @@ pub struct Task {
 }
 
 fn log_sink(cfg: Arc<Config>, task_id: Uuid) -> impl Sink<String, Error = anyhow::Error> + Send {
-    futures_util::sink::unfold((), move |_, line| {
+    futures_util::sink::unfold((), move |_, lines| {
         let cfg = cfg.clone();
         async move {
             api::<_, ()>(
                 &cfg,
                 Method::POST,
                 &format!("api/v1/tasks/{}/logs", task_id),
-                &serde_json::json!({ "lines": [line] }),
+                &serde_json::json!({ "lines": lines }),
             )
             .await
         }
@@ -112,7 +112,12 @@ async fn try_run_task(cfg: Arc<Config>, task: &Task) -> anyhow::Result<()> {
         FramedRead::new(child.stderr.take().unwrap(), LinesCodec::new()),
     );
 
-    lines.map_err(anyhow::Error::from).forward(logger).await?;
+    lines
+        .ready_chunks(1024) // TODO switch to try_ready_chunks
+        .map(|v| v.into_iter().collect::<Result<_, _>>())
+        .map_err(anyhow::Error::from)
+        .forward(logger)
+        .await?;
     let exit_status = child.wait().await?;
 
     if exit_status.success() {