Implement Container Task API

Signed-off-by: Guvenc Gulce <[email protected]>

Author: guvenc

Cargo.lock

@@ -4,6 +4,7 @@ members = [
     "feos/services/vm-service",
     "feos/services/host-service",
     "feos/services/image-service",
+    "feos/services/task-service",
     "cli",
     "feos/proto",
     "feos/utils",

Cargo.toml

@@ -16,6 +16,7 @@ feos-utils = { path = "utils" }
 vm-service = { path = "services/vm-service" }
 host-service = { path = "services/host-service" }
 image-service = { path = "services/image-service" }
+task-service = { path = "services/task-service" }
 feos-proto = { workspace = true }
 
 # Workspace dependencies

feos/Cargo.toml

@@ -17,4 +17,4 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
             &[proto_dir],
         )?;
     Ok(())
-}
+}

feos/proto/build.rs

@@ -10,3 +10,6 @@ pub mod host_service {
 pub mod image_service {
     tonic::include_proto!("feos.image.vmm.api.v1");
 }
+pub mod task_service {
+    tonic::include_proto!("feos.task.v1");
+}

feos/proto/src/lib.rs

@@ -87,9 +87,7 @@ impl FileStore {
         fs::create_dir_all(final_dir).await?;
 
         // Handle both single-blob rootfs (for VMs) and layered rootfs (for containers)
-        if image_data.layers.len() == 1
-            && image_ref.contains("cloud-hypervisor")
-        {
+        if image_data.layers.len() == 1 && image_ref.contains("cloud-hypervisor") {
             // Assuming this is a single rootfs blob for a VM
             let final_disk_path = final_dir.join("disk.image");
             fs::write(final_disk_path, &image_data.layers[0]).await?;
@@ -163,4 +161,4 @@ impl FileStore {
         );
         store
     }
-}
+}

feos/services/image-service/src/filestore.rs

@@ -95,4 +95,4 @@ pub enum FileCommand {
     ScanExistingImages {
         responder: oneshot::Sender<HashMap<String, ImageInfo>>,
     },
-}
+}

feos/services/image-service/src/lib.rs

@@ -9,9 +9,7 @@ use feos_proto::image_service::{
     PullImageResponse,
 };
 use log::{error, info, warn};
-use oci_distribution::{
-    client::ClientConfig, manifest, secrets::RegistryAuth, Client, Reference,
-};
+use oci_distribution::{client::ClientConfig, manifest, secrets::RegistryAuth, Client, Reference};
 use std::collections::HashMap;
 use tokio::sync::{broadcast, mpsc, oneshot};
 use tonic::Status;
@@ -257,9 +255,7 @@ async fn pull_oci_data(image_ref: &str) -> Result<PulledImageData, ImageServiceE
     let auth = &RegistryAuth::Anonymous;
 
     info!("ImagePuller: pulling manifest and config for {image_ref}");
-    let (manifest, _, _) = client
-        .pull_manifest_and_config(&reference, auth)
-        .await?;
+    let (manifest, _, _) = client.pull_manifest_and_config(&reference, auth).await?;
 
     let mut config_data = Vec::new();
     client
@@ -289,10 +285,7 @@ async fn pull_oci_data(image_ref: &str) -> Result<PulledImageData, ImageServiceE
         client
             .pull_blob(&reference, &layer, &mut layer_data)
             .await?;
-        info!(
-            "ImagePuller: pulled layer blob {} bytes",
-            layer_data.len()
-        );
+        info!("ImagePuller: pulled layer blob {} bytes", layer_data.len());
         layers.push(layer_data);
     }
 
@@ -411,4 +404,4 @@ pub async fn watch_image_status_stream(
             }
         }
     }
-}
+}

feos/services/image-service/src/worker.rs

@@ -0,0 +1,15 @@
+[package]
+name = "task-service"
+version.workspace = true
+edition.workspace = true
+
+[dependencies]
+feos-proto = { workspace = true }
+tokio = { workspace = true }
+tokio-stream = { workspace = true }
+tonic = { workspace = true }
+anyhow = { workspace = true }
+log = { workspace = true }
+prost = { workspace = true }
+thiserror = { workspace = true }
+nix = { workspace = true, features = ["signal", "process"] }

feos/services/task-service/Cargo.toml

@@ -0,0 +1,116 @@
+use crate::error::TaskError;
+use crate::Command;
+use feos_proto::task_service::{
+    task_service_server::TaskService, CreateRequest, CreateResponse, DeleteRequest, DeleteResponse,
+    KillRequest, KillResponse, StartRequest, StartResponse, WaitRequest, WaitResponse,
+};
+use log::info;
+use tokio::sync::{mpsc, oneshot};
+use tonic::{Request, Response, Status};
+
+pub struct TaskApiHandler {
+    dispatcher_tx: mpsc::Sender<Command>,
+}
+
+impl TaskApiHandler {
+    pub fn new(dispatcher_tx: mpsc::Sender<Command>) -> Self {
+        Self { dispatcher_tx }
+    }
+}
+
+/// Helper function to create a command, send it to the dispatcher, and await the response.
+async fn dispatch_and_wait<T, F>(
+    dispatcher: &mpsc::Sender<Command>,
+    command_constructor: F,
+) -> Result<Response<T>, Status>
+where
+    F: FnOnce(oneshot::Sender<Result<T, TaskError>>) -> Command,
+{
+    let (resp_tx, resp_rx) = oneshot::channel();
+    let cmd = command_constructor(resp_tx);
+
+    dispatcher
+        .send(cmd)
+        .await
+        .map_err(|e| Status::internal(format!("Failed to send command to dispatcher: {}", e)))?;
+
+    match resp_rx.await {
+        Ok(Ok(result)) => Ok(Response::new(result)),
+        Ok(Err(e)) => Err(e.into()),
+        Err(_) => Err(Status::internal(
+            "Dispatcher task dropped response channel.",
+        )),
+    }
+}
+
+#[tonic::async_trait]
+impl TaskService for TaskApiHandler {
+    async fn create(
+        &self,
+        request: Request<CreateRequest>,
+    ) -> Result<Response<CreateResponse>, Status> {
+        info!(
+            "API: Received Create request for {}",
+            request.get_ref().container_id
+        );
+        dispatch_and_wait(&self.dispatcher_tx, |responder| Command::Create {
+            req: request.into_inner(),
+            responder,
+        })
+        .await
+    }
+
+    async fn start(
+        &self,
+        request: Request<StartRequest>,
+    ) -> Result<Response<StartResponse>, Status> {
+        info!(
+            "API: Received Start request for {}",
+            request.get_ref().container_id
+        );
+        dispatch_and_wait(&self.dispatcher_tx, |responder| Command::Start {
+            req: request.into_inner(),
+            responder,
+        })
+        .await
+    }
+
+    async fn kill(&self, request: Request<KillRequest>) -> Result<Response<KillResponse>, Status> {
+        info!(
+            "API: Received Kill request for {}",
+            request.get_ref().container_id
+        );
+        dispatch_and_wait(&self.dispatcher_tx, |responder| Command::Kill {
+            req: request.into_inner(),
+            responder,
+        })
+        .await
+    }
+
+    async fn delete(
+        &self,
+        request: Request<DeleteRequest>,
+    ) -> Result<Response<DeleteResponse>, Status> {
+        info!(
+            "API: Received Delete request for {}",
+            request.get_ref().container_id
+        );
+        dispatch_and_wait(&self.dispatcher_tx, |responder| Command::Delete {
+            req: request.into_inner(),
+            responder,
+        })
+        .await
+    }
+
+    async fn wait(&self, request: Request<WaitRequest>) -> Result<Response<WaitResponse>, Status> {
+        info!(
+            "API: Received Wait request for {}",
+            request.get_ref().container_id
+        );
+        dispatch_and_wait(&self.dispatcher_tx, |responder| Command::Wait {
+            req: request.into_inner(),
+            responder,
+        })
+        .await
+    }
+}