feat: oci compliance work (#85)

* chore: rework oci crate to be more composable

* feat: image pull is now internally explicit

* feat: utilize vfs for assembling oci images

* feat: rework oci to preserve permissions via a vfs

crates/daemon/src/control.rs

@@ -1,5 +1,3 @@
-use std::{pin::Pin, str::FromStr};
-
 use async_stream::try_stream;
 use futures::Stream;
 use krata::{
@@ -8,19 +6,26 @@ use krata::{
         IdmMetricsRequest,
     },
     v1::{
-        common::{Guest, GuestState, GuestStatus},
+        common::{Guest, GuestOciImageFormat, GuestState, GuestStatus},
         control::{
             control_service_server::ControlService, ConsoleDataReply, ConsoleDataRequest,
             CreateGuestReply, CreateGuestRequest, DestroyGuestReply, DestroyGuestRequest,
-            ListGuestsReply, ListGuestsRequest, ReadGuestMetricsReply, ReadGuestMetricsRequest,
-            ResolveGuestReply, ResolveGuestRequest, SnoopIdmReply, SnoopIdmRequest,
-            WatchEventsReply, WatchEventsRequest,
+            ListGuestsReply, ListGuestsRequest, PullImageReply, PullImageRequest,
+            ReadGuestMetricsReply, ReadGuestMetricsRequest, ResolveGuestReply, ResolveGuestRequest,
+            SnoopIdmReply, SnoopIdmRequest, WatchEventsReply, WatchEventsRequest,
         },
     },
 };
+use krataoci::{
+    name::ImageName,
+    packer::{service::OciPackerService, OciImagePacked, OciPackedFormat},
+    progress::{OciProgress, OciProgressContext},
+};
+use std::{pin::Pin, str::FromStr};
 use tokio::{
     select,
     sync::mpsc::{channel, Sender},
+    task::JoinError,
 };
 use tokio_stream::StreamExt;
 use tonic::{Request, Response, Status, Streaming};
@@ -28,7 +33,7 @@ use uuid::Uuid;
 
 use crate::{
     console::DaemonConsoleHandle, db::GuestStore, event::DaemonEventContext, idm::DaemonIdmHandle,
-    metrics::idm_metric_to_api,
+    metrics::idm_metric_to_api, oci::convert_oci_progress,
 };
 
 pub struct ApiError {
@@ -50,21 +55,23 @@ impl From<ApiError> for Status {
 }
 
 #[derive(Clone)]
-pub struct RuntimeControlService {
+pub struct DaemonControlService {
     events: DaemonEventContext,
     console: DaemonConsoleHandle,
     idm: DaemonIdmHandle,
     guests: GuestStore,
     guest_reconciler_notify: Sender<Uuid>,
+    packer: OciPackerService,
 }
 
-impl RuntimeControlService {
+impl DaemonControlService {
     pub fn new(
         events: DaemonEventContext,
         console: DaemonConsoleHandle,
         idm: DaemonIdmHandle,
         guests: GuestStore,
         guest_reconciler_notify: Sender<Uuid>,
+        packer: OciPackerService,
     ) -> Self {
         Self {
             events,
@@ -72,6 +79,7 @@ impl RuntimeControlService {
             idm,
             guests,
             guest_reconciler_notify,
+            packer,
         }
     }
 }
@@ -81,11 +89,19 @@ enum ConsoleDataSelect {
     Write(Option<Result<ConsoleDataRequest, tonic::Status>>),
 }
 
+enum PullImageSelect {
+    Progress(usize),
+    Completed(Result<Result<OciImagePacked, anyhow::Error>, JoinError>),
+}
+
 #[tonic::async_trait]
-impl ControlService for RuntimeControlService {
+impl ControlService for DaemonControlService {
     type ConsoleDataStream =
         Pin<Box<dyn Stream<Item = Result<ConsoleDataReply, Status>> + Send + 'static>>;
 
+    type PullImageStream =
+        Pin<Box<dyn Stream<Item = Result<PullImageReply, Status>> + Send + 'static>>;
+
     type WatchEventsStream =
         Pin<Box<dyn Stream<Item = Result<WatchEventsReply, Status>> + Send + 'static>>;
 
@@ -337,6 +353,71 @@ impl ControlService for RuntimeControlService {
         Ok(Response::new(reply))
     }
 
+    async fn pull_image(
+        &self,
+        request: Request<PullImageRequest>,
+    ) -> Result<Response<Self::PullImageStream>, Status> {
+        let request = request.into_inner();
+        let name = ImageName::parse(&request.image).map_err(|err| ApiError {
+            message: err.to_string(),
+        })?;
+        let format = match request.format() {
+            GuestOciImageFormat::Unknown => OciPackedFormat::Squashfs,
+            GuestOciImageFormat::Squashfs => OciPackedFormat::Squashfs,
+            GuestOciImageFormat::Erofs => OciPackedFormat::Erofs,
+        };
+        let (sender, mut receiver) = channel::<OciProgress>(100);
+        let context = OciProgressContext::new(sender);
+
+        let our_packer = self.packer.clone();
+
+        let output = try_stream! {
+            let mut task = tokio::task::spawn(async move {
+                our_packer.request(name, format, context).await
+            });
+            loop {
+                let mut progresses = Vec::new();
+                let what = select! {
+                    x = receiver.recv_many(&mut progresses, 10) => PullImageSelect::Progress(x),
+                    x = &mut task => PullImageSelect::Completed(x),
+                };
+                match what {
+                    PullImageSelect::Progress(count) => {
+                        if count > 0 {
+                            let progress = progresses.remove(progresses.len() - 1);
+                            let reply = PullImageReply {
+                                progress: Some(convert_oci_progress(progress)),
+                                digest: String::new(),
+                                format: GuestOciImageFormat::Unknown.into(),
+                            };
+                            yield reply;
+                        }
+                    },
+
+                    PullImageSelect::Completed(result) => {
+                        let result = result.map_err(|err| ApiError {
+                            message: err.to_string(),
+                        })?;
+                        let packed = result.map_err(|err| ApiError {
+                            message: err.to_string(),
+                        })?;
+                        let reply = PullImageReply {
+                            progress: None,
+                            digest: packed.digest,
+                            format: match packed.format {
+                                OciPackedFormat::Squashfs => GuestOciImageFormat::Squashfs.into(),
+                                OciPackedFormat::Erofs => GuestOciImageFormat::Erofs.into(),
+                            },
+                        };
+                        yield reply;
+                        break;
+                    },
+                }
+            }
+        };
+        Ok(Response::new(Box::pin(output) as Self::PullImageStream))
+    }
+
     async fn watch_events(
         &self,
         request: Request<WatchEventsRequest>,

crates/daemon/src/event.rs

@@ -9,7 +9,6 @@ use krata::{
     idm::protocol::{idm_event::Event, IdmEvent},
     v1::common::{GuestExitInfo, GuestState, GuestStatus},
 };
-use krataoci::progress::OciProgress;
 use log::{error, warn};
 use tokio::{
     select,
@@ -22,7 +21,7 @@ use tokio::{
 };
 use uuid::Uuid;
 
-use crate::{db::GuestStore, idm::DaemonIdmHandle, oci::convert_oci_progress};
+use crate::{db::GuestStore, idm::DaemonIdmHandle};
 
 pub type DaemonEvent = krata::v1::control::watch_events_reply::Event;
 
@@ -53,8 +52,7 @@ pub struct DaemonEventGenerator {
     idms: HashMap<u32, (Uuid, JoinHandle<()>)>,
     idm_sender: Sender<(u32, IdmEvent)>,
     idm_receiver: Receiver<(u32, IdmEvent)>,
-    oci_progress_receiver: broadcast::Receiver<OciProgress>,
-    event_sender: broadcast::Sender<DaemonEvent>,
+    _event_sender: broadcast::Sender<DaemonEvent>,
 }
 
 impl DaemonEventGenerator {
@@ -62,7 +60,6 @@ impl DaemonEventGenerator {
         guests: GuestStore,
         guest_reconciler_notify: Sender<Uuid>,
         idm: DaemonIdmHandle,
-        oci_progress_receiver: broadcast::Receiver<OciProgress>,
     ) -> Result<(DaemonEventContext, DaemonEventGenerator)> {
         let (sender, _) = broadcast::channel(EVENT_CHANNEL_QUEUE_LEN);
         let (idm_sender, idm_receiver) = channel(IDM_EVENT_CHANNEL_QUEUE_LEN);
@@ -74,55 +71,53 @@ impl DaemonEventGenerator {
             idms: HashMap::new(),
             idm_sender,
             idm_receiver,
-            oci_progress_receiver,
-            event_sender: sender.clone(),
+            _event_sender: sender.clone(),
         };
         let context = DaemonEventContext { sender };
         Ok((context, generator))
     }
 
     async fn handle_feed_event(&mut self, event: &DaemonEvent) -> Result<()> {
-        if let DaemonEvent::GuestChanged(changed) = event {
-            let Some(ref guest) = changed.guest else {
-                return Ok(());
-            };
+        let DaemonEvent::GuestChanged(changed) = event;
+        let Some(ref guest) = changed.guest else {
+            return Ok(());
+        };
 
-            let Some(ref state) = guest.state else {
-                return Ok(());
-            };
+        let Some(ref state) = guest.state else {
+            return Ok(());
+        };
 
-            let status = state.status();
-            let id = Uuid::from_str(&guest.id)?;
-            let domid = state.domid;
-            match status {
-                GuestStatus::Started => {
-                    if let Entry::Vacant(e) = self.idms.entry(domid) {
-                        let client = self.idm.client(domid).await?;
-                        let mut receiver = client.subscribe().await?;
-                        let sender = self.idm_sender.clone();
-                        let task = tokio::task::spawn(async move {
-                            loop {
-                                let Ok(event) = receiver.recv().await else {
-                                    break;
-                                };
+        let status = state.status();
+        let id = Uuid::from_str(&guest.id)?;
+        let domid = state.domid;
+        match status {
+            GuestStatus::Started => {
+                if let Entry::Vacant(e) = self.idms.entry(domid) {
+                    let client = self.idm.client(domid).await?;
+                    let mut receiver = client.subscribe().await?;
+                    let sender = self.idm_sender.clone();
+                    let task = tokio::task::spawn(async move {
+                        loop {
+                            let Ok(event) = receiver.recv().await else {
+                                break;
+                            };
 
-                                if let Err(error) = sender.send((domid, event)).await {
-                                    warn!("unable to deliver idm event: {}", error);
-                                }
+                            if let Err(error) = sender.send((domid, event)).await {
+                                warn!("unable to deliver idm event: {}", error);
                             }
-                        });
-                        e.insert((id, task));
-                    }
+                        }
+                    });
+                    e.insert((id, task));
                 }
-
-                GuestStatus::Destroyed => {
-                    if let Some((_, handle)) = self.idms.remove(&domid) {
-                        handle.abort();
-                    }
-                }
-
-                _ => {}
             }
+
+            GuestStatus::Destroyed => {
+                if let Some((_, handle)) = self.idms.remove(&domid) {
+                    handle.abort();
+                }
+            }
+
+            _ => {}
         }
         Ok(())
     }
@@ -150,17 +145,6 @@ impl DaemonEventGenerator {
         Ok(())
     }
 
-    async fn handle_oci_progress_event(&mut self, progress: OciProgress) -> Result<()> {
-        let Some(_) = Uuid::from_str(&progress.id).ok() else {
-            return Ok(());
-        };
-
-        let event = convert_oci_progress(progress);
-        self.event_sender.send(DaemonEvent::OciProgress(event))?;
-
-        Ok(())
-    }
-
     async fn evaluate(&mut self) -> Result<()> {
         select! {
             x = self.idm_receiver.recv() => match x {
@@ -182,14 +166,6 @@ impl DaemonEventGenerator {
                     Err(error.into())
                 }
             },
-            x = self.oci_progress_receiver.recv() => match x {
-                Ok(event) => {
-                    self.handle_oci_progress_event(event).await
-                },
-                Err(error) => {
-                    Err(error.into())
-                }
-            }
         }
     }
 

crates/daemon/src/lib.rs

@@ -2,21 +2,19 @@ use std::{net::SocketAddr, path::PathBuf, str::FromStr};
 
 use anyhow::Result;
 use console::{DaemonConsole, DaemonConsoleHandle};
-use control::RuntimeControlService;
+use control::DaemonControlService;
 use db::GuestStore;
 use event::{DaemonEventContext, DaemonEventGenerator};
 use idm::{DaemonIdm, DaemonIdmHandle};
 use krata::{dial::ControlDialAddress, v1::control::control_service_server::ControlServiceServer};
-use krataoci::progress::OciProgressContext;
+use krataoci::{packer::service::OciPackerService, registry::OciPlatform};
 use kratart::Runtime;
 use log::info;
 use reconcile::guest::GuestReconciler;
 use tokio::{
+    fs,
     net::UnixListener,
-    sync::{
-        broadcast,
-        mpsc::{channel, Sender},
-    },
+    sync::mpsc::{channel, Sender},
     task::JoinHandle,
 };
 use tokio_stream::wrappers::UnixListenerStream;
@@ -41,17 +39,21 @@ pub struct Daemon {
     generator_task: JoinHandle<()>,
     idm: DaemonIdmHandle,
     console: DaemonConsoleHandle,
+    packer: OciPackerService,
 }
 
 const GUEST_RECONCILER_QUEUE_LEN: usize = 1000;
-const OCI_PROGRESS_QUEUE_LEN: usize = 1000;
 
 impl Daemon {
     pub async fn new(store: String) -> Result<Self> {
-        let (oci_progress_sender, oci_progress_receiver) =
-            broadcast::channel(OCI_PROGRESS_QUEUE_LEN);
-        let runtime =
-            Runtime::new(OciProgressContext::new(oci_progress_sender), store.clone()).await?;
+        let mut image_cache_dir = PathBuf::from(store.clone());
+        image_cache_dir.push("cache");
+        image_cache_dir.push("image");
+        fs::create_dir_all(&image_cache_dir).await?;
+
+        let packer = OciPackerService::new(None, &image_cache_dir, OciPlatform::current())?;
+
+        let runtime = Runtime::new(store.clone()).await?;
         let guests_db_path = format!("{}/guests.db", store);
         let guests = GuestStore::open(&PathBuf::from(guests_db_path))?;
         let (guest_reconciler_notify, guest_reconciler_receiver) =
@@ -60,23 +62,21 @@ impl Daemon {
         let idm = idm.launch().await?;
         let console = DaemonConsole::new().await?;
         let console = console.launch().await?;
-        let (events, generator) = DaemonEventGenerator::new(
-            guests.clone(),
-            guest_reconciler_notify.clone(),
-            idm.clone(),
-            oci_progress_receiver,
-        )
-        .await?;
+        let (events, generator) =
+            DaemonEventGenerator::new(guests.clone(), guest_reconciler_notify.clone(), idm.clone())
+                .await?;
         let runtime_for_reconciler = runtime.dupe().await?;
         let guest_reconciler = GuestReconciler::new(
             guests.clone(),
             events.clone(),
             runtime_for_reconciler,
+            packer.clone(),
             guest_reconciler_notify.clone(),
         )?;
 
         let guest_reconciler_task = guest_reconciler.launch(guest_reconciler_receiver).await?;
         let generator_task = generator.launch().await?;
+
         Ok(Self {
             store,
             guests,
@@ -86,16 +86,18 @@ impl Daemon {
             generator_task,
             idm,
             console,
+            packer,
         })
     }
 
     pub async fn listen(&mut self, addr: ControlDialAddress) -> Result<()> {
-        let control_service = RuntimeControlService::new(
+        let control_service = DaemonControlService::new(
             self.events.clone(),
             self.console.clone(),
             self.idm.clone(),
             self.guests.clone(),
             self.guest_reconciler_notify.clone(),
+            self.packer.clone(),
         );
 
         let mut server = Server::builder();
@@ -121,7 +123,7 @@ impl Daemon {
             ControlDialAddress::UnixSocket { path } => {
                 let path = PathBuf::from(path);
                 if path.exists() {
-                    tokio::fs::remove_file(&path).await?;
+                    fs::remove_file(&path).await?;
                 }
                 let listener = UnixListener::bind(path)?;
                 let stream = UnixListenerStream::new(listener);

crates/daemon/src/oci.rs

@@ -1,17 +1,17 @@
 use krata::v1::control::{
-    OciProgressEvent, OciProgressEventLayer, OciProgressEventLayerPhase, OciProgressEventPhase,
+    PullImageProgress, PullImageProgressLayer, PullImageProgressLayerPhase, PullImageProgressPhase,
 };
 use krataoci::progress::{OciProgress, OciProgressLayer, OciProgressLayerPhase, OciProgressPhase};
 
-fn convert_oci_layer_progress(layer: OciProgressLayer) -> OciProgressEventLayer {
-    OciProgressEventLayer {
+fn convert_oci_layer_progress(layer: OciProgressLayer) -> PullImageProgressLayer {
+    PullImageProgressLayer {
         id: layer.id,
         phase: match layer.phase {
-            OciProgressLayerPhase::Waiting => OciProgressEventLayerPhase::Waiting,
-            OciProgressLayerPhase::Downloading => OciProgressEventLayerPhase::Downloading,
-            OciProgressLayerPhase::Downloaded => OciProgressEventLayerPhase::Downloaded,
-            OciProgressLayerPhase::Extracting => OciProgressEventLayerPhase::Extracting,
-            OciProgressLayerPhase::Extracted => OciProgressEventLayerPhase::Extracted,
+            OciProgressLayerPhase::Waiting => PullImageProgressLayerPhase::Waiting,
+            OciProgressLayerPhase::Downloading => PullImageProgressLayerPhase::Downloading,
+            OciProgressLayerPhase::Downloaded => PullImageProgressLayerPhase::Downloaded,
+            OciProgressLayerPhase::Extracting => PullImageProgressLayerPhase::Extracting,
+            OciProgressLayerPhase::Extracted => PullImageProgressLayerPhase::Extracted,
         }
         .into(),
         value: layer.value,
@@ -19,16 +19,15 @@ fn convert_oci_layer_progress(layer: OciProgressLayer) -> OciProgressEventLayer
     }
 }
 
-pub fn convert_oci_progress(oci: OciProgress) -> OciProgressEvent {
-    OciProgressEvent {
-        guest_id: oci.id,
+pub fn convert_oci_progress(oci: OciProgress) -> PullImageProgress {
+    PullImageProgress {
         phase: match oci.phase {
-            OciProgressPhase::Resolving => OciProgressEventPhase::Resolving,
-            OciProgressPhase::Resolved => OciProgressEventPhase::Resolved,
-            OciProgressPhase::ConfigAcquire => OciProgressEventPhase::ConfigAcquire,
-            OciProgressPhase::LayerAcquire => OciProgressEventPhase::LayerAcquire,
-            OciProgressPhase::Packing => OciProgressEventPhase::Packing,
-            OciProgressPhase::Complete => OciProgressEventPhase::Complete,
+            OciProgressPhase::Resolving => PullImageProgressPhase::Resolving,
+            OciProgressPhase::Resolved => PullImageProgressPhase::Resolved,
+            OciProgressPhase::ConfigAcquire => PullImageProgressPhase::ConfigAcquire,
+            OciProgressPhase::LayerAcquire => PullImageProgressPhase::LayerAcquire,
+            OciProgressPhase::Packing => PullImageProgressPhase::Packing,
+            OciProgressPhase::Complete => PullImageProgressPhase::Complete,
         }
         .into(),
         layers: oci

crates/daemon/src/reconcile/guest.rs

@@ -9,10 +9,11 @@ use krata::launchcfg::LaunchPackedFormat;
 use krata::v1::{
     common::{
         guest_image_spec::Image, Guest, GuestErrorInfo, GuestExitInfo, GuestNetworkState,
-        GuestState, GuestStatus,
+        GuestOciImageFormat, GuestState, GuestStatus,
     },
     control::GuestChangedEvent,
 };
+use krataoci::packer::{service::OciPackerService, OciPackedFormat};
 use kratart::{launch::GuestLaunchRequest, GuestInfo, Runtime};
 use log::{error, info, trace, warn};
 use tokio::{
@@ -55,6 +56,7 @@ pub struct GuestReconciler {
     guests: GuestStore,
     events: DaemonEventContext,
     runtime: Runtime,
+    packer: OciPackerService,
     tasks: Arc<Mutex<HashMap<Uuid, GuestReconcilerEntry>>>,
     guest_reconciler_notify: Sender<Uuid>,
     reconcile_lock: Arc<RwLock<()>>,
@@ -65,12 +67,14 @@ impl GuestReconciler {
         guests: GuestStore,
         events: DaemonEventContext,
         runtime: Runtime,
+        packer: OciPackerService,
         guest_reconciler_notify: Sender<Uuid>,
     ) -> Result<Self> {
         Ok(Self {
             guests,
             events,
             runtime,
+            packer,
             tasks: Arc::new(Mutex::new(HashMap::new())),
             guest_reconciler_notify,
             reconcile_lock: Arc::new(RwLock::with_max_readers((), PARALLEL_LIMIT)),
@@ -233,9 +237,27 @@ impl GuestReconciler {
                 return Err(anyhow!("oci spec not specified"));
             }
         };
-
         let task = spec.task.as_ref().cloned().unwrap_or_default();
 
+        let image = self
+            .packer
+            .recall(
+                &oci.digest,
+                match oci.format() {
+                    GuestOciImageFormat::Unknown => OciPackedFormat::Squashfs,
+                    GuestOciImageFormat::Squashfs => OciPackedFormat::Squashfs,
+                    GuestOciImageFormat::Erofs => OciPackedFormat::Erofs,
+                },
+            )
+            .await?;
+
+        let Some(image) = image else {
+            return Err(anyhow!(
+                "image {} in the requested format did not exist",
+                oci.digest
+            ));
+        };
+
         let info = self
             .runtime
             .launch(GuestLaunchRequest {
@@ -244,9 +266,9 @@ impl GuestReconciler {
                 name: if spec.name.is_empty() {
                     None
                 } else {
-                    Some(&spec.name)
+                    Some(spec.name.clone())
                 },
-                image: &oci.image,
+                image,
                 vcpus: spec.vcpus,
                 mem: spec.mem,
                 env: task