agent: Introduce ImagePullHandler to support IMAGE_GUEST_PULL volume

As we do not employ a forked containerd in confidential-containers, we utilize the KataVirtualVolume
which storing the image information as an integral part of `CreateContainer`.
Within this process, we store the image information in rootfs.storage and pass this image url through `CreateContainerRequest`.
This approach distinguishes itself from the use of `PullImageRequest`, as rootfs.storage is already set and initialized at this stage.
To maintain clarity and avoid any need for modification to the `OverlayfsHandler`,we introduce the `ImagePullHandler`.
This dedicated handler is responsible for orchestrating the image-pulling logic within the guest environment.
This logic encompasses tasks such as calling the image-rs to download and unpack the image into `/run/kata-containers/{container_id}/images`,
followed by a bind mount to `/run/kata-containers/{container_id}`.

Signed-off-by: ChengyuZhu6 <chengyu.zhu@intel.com>

src/agent/src/storage/image_pull_handler.rs

@@ -0,0 +1,102 @@
+// Copyright (c) 2023 Intel Corporation
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+
+use anyhow::{anyhow, Result};
+use kata_types::mount::KATA_VIRTUAL_VOLUME_IMAGE_GUEST_PULL;
+use kata_types::mount::{ImagePullVolume, StorageDevice};
+use protocols::agent::Storage;
+use std::sync::Arc;
+use tracing::instrument;
+
+use crate::image;
+use crate::storage::{StorageContext, StorageHandler};
+
+use super::{common_storage_handler, new_device};
+
+#[derive(Debug)]
+pub struct ImagePullHandler {}
+
+impl ImagePullHandler {
+    fn get_image_info(storage: &Storage) -> Result<ImagePullVolume> {
+        for option in storage.driver_options.iter() {
+            if let Some((key, value)) = option.split_once('=') {
+                if key == KATA_VIRTUAL_VOLUME_IMAGE_GUEST_PULL {
+                    let imagepull_volume: ImagePullVolume = serde_json::from_str(value)?;
+                    return Ok(imagepull_volume);
+                }
+            }
+        }
+        Err(anyhow!("missing Image information for ImagePull volume"))
+    }
+}
+
+#[async_trait::async_trait]
+impl StorageHandler for ImagePullHandler {
+    #[instrument]
+    async fn create_device(
+        &self,
+        mut storage: Storage,
+        ctx: &mut StorageContext,
+    ) -> Result<Arc<dyn StorageDevice>> {
+        //Currently the image metadata is not used to pulling image in the guest.
+        let image_pull_volume = Self::get_image_info(&storage)?;
+        debug!(ctx.logger, "image_pull_volume = {:?}", image_pull_volume);
+        let image_name = storage.source();
+        debug!(ctx.logger, "image_name = {:?}", image_name);
+
+        let cid = ctx
+            .cid
+            .clone()
+            .ok_or_else(|| anyhow!("failed to get container id"))?;
+        let image_service = image::ImageService::singleton().await?;
+        let bundle_path = image_service
+            .pull_image(image_name, &cid, &image_pull_volume.metadata)
+            .await?;
+
+        storage.source = bundle_path;
+        storage.options = vec!["bind".to_string(), "ro".to_string()];
+
+        common_storage_handler(ctx.logger, &storage)?;
+
+        new_device(storage.mount_point)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::collections::HashMap;
+
+    use kata_types::mount::{ImagePullVolume, KATA_VIRTUAL_VOLUME_IMAGE_GUEST_PULL};
+    use protocols::agent::Storage;
+
+    use crate::storage::image_pull_handler::ImagePullHandler;
+
+    #[test]
+    fn test_get_image_info() {
+        let mut res = HashMap::new();
+        res.insert("key1".to_string(), "value1".to_string());
+        res.insert("key2".to_string(), "value2".to_string());
+
+        let image_pull = ImagePullVolume {
+            metadata: res.clone(),
+        };
+
+        let image_pull_str = serde_json::to_string(&image_pull);
+        assert!(image_pull_str.is_ok());
+
+        let storage = Storage {
+            driver: KATA_VIRTUAL_VOLUME_IMAGE_GUEST_PULL.to_string(),
+            driver_options: vec![format!("image_guest_pull={}", image_pull_str.ok().unwrap())],
+            ..Default::default()
+        };
+
+        match ImagePullHandler::get_image_info(&storage) {
+            Ok(image_info) => {
+                assert_eq!(image_info.metadata, res);
+            }
+            Err(e) => panic!("err = {}", e),
+        }
+    }
+}

src/agent/src/storage/mod.rs

@@ -12,7 +12,10 @@ use std::sync::Arc;
 
 use anyhow::{anyhow, Context, Result};
 use kata_sys_util::mount::{create_mount_destination, parse_mount_options};
-use kata_types::mount::{StorageDevice, StorageHandlerManager, KATA_SHAREDFS_GUEST_PREMOUNT_TAG};
+use kata_types::mount::{
+    StorageDevice, StorageHandlerManager, KATA_SHAREDFS_GUEST_PREMOUNT_TAG,
+    KATA_VIRTUAL_VOLUME_IMAGE_GUEST_PULL,
+};
 use nix::unistd::{Gid, Uid};
 use protocols::agent::Storage;
 use protocols::types::FSGroupChangePolicy;
@@ -24,6 +27,7 @@ use self::bind_watcher_handler::BindWatcherHandler;
 use self::block_handler::{PmemHandler, ScsiHandler, VirtioBlkMmioHandler, VirtioBlkPciHandler};
 use self::ephemeral_handler::EphemeralHandler;
 use self::fs_handler::{OverlayfsHandler, Virtio9pHandler, VirtioFsHandler};
+use self::image_pull_handler::ImagePullHandler;
 use self::local_handler::LocalHandler;
 use crate::device::{
     DRIVER_9P_TYPE, DRIVER_BLK_MMIO_TYPE, DRIVER_BLK_PCI_TYPE, DRIVER_EPHEMERAL_TYPE,
@@ -39,6 +43,7 @@ mod bind_watcher_handler;
 mod block_handler;
 mod ephemeral_handler;
 mod fs_handler;
+mod image_pull_handler;
 mod local_handler;
 
 const RW_MASK: u32 = 0o660;
@@ -145,6 +150,7 @@ lazy_static! {
         manager.add_handler(DRIVER_SCSI_TYPE, Arc::new(ScsiHandler{})).unwrap();
         manager.add_handler(DRIVER_VIRTIOFS_TYPE, Arc::new(VirtioFsHandler{})).unwrap();
         manager.add_handler(DRIVER_WATCHABLE_BIND_TYPE, Arc::new(BindWatcherHandler{})).unwrap();
+        manager.add_handler(KATA_VIRTUAL_VOLUME_IMAGE_GUEST_PULL, Arc::new(ImagePullHandler{})).unwrap();
         manager
     };
 }