oci: add support for pulling decompressed tar

Check the media_type on the descriptor before attempting gzip
decompression.  If the media type indicates an uncompressed layer, then
consume the stream directly.  Add zstd support as well.

Unfortunately the code is a bit awkward: split_async() is generic, so we
need to make a separate call for each potential underlying AsyncBufRead
impl.

Signed-off-by: Allison Karlitskaya <[email protected]>

Author: allisonkarlitskaya

Cargo.toml

@@ -18,7 +18,7 @@ rhel9 = ['pre-6.15']
 
 [dependencies]
 anyhow = { version = "1.0.97", default-features = false }
-async-compression = { version = "0.4.22", default-features = false, features = ["tokio", "gzip"] }
+async-compression = { version = "0.4.22", default-features = false, features = ["tokio", "zstd", "gzip"] }
 clap = { version = "4.5.32", default-features = false, features = ["std", "help", "usage", "derive"] }
 containers-image-proxy = "0.7.0"
 env_logger = "0.11.7"

src/oci/mod.rs

@@ -4,10 +4,10 @@ pub mod tar;
 use std::{collections::HashMap, io::Read, iter::zip, path::Path};
 
 use anyhow::{bail, ensure, Context, Result};
-use async_compression::tokio::bufread::GzipDecoder;
+use async_compression::tokio::bufread::{GzipDecoder, ZstdDecoder};
 use containers_image_proxy::{ImageProxy, ImageProxyConfig, OpenedImage};
 use indicatif::{MultiProgress, ProgressBar, ProgressStyle};
-use oci_spec::image::{Descriptor, ImageConfiguration, ImageManifest};
+use oci_spec::image::{Descriptor, ImageConfiguration, ImageManifest, MediaType};
 use sha2::{Digest, Sha256};
 
 use crate::{
@@ -101,9 +101,20 @@ impl<'repo> ImageOp<'repo> {
             let progress = bar.wrap_async_read(blob_reader);
             self.progress
                 .println(format!("Fetching layer {}", hex::encode(layer_sha256)))?;
-            let decoder = GzipDecoder::new(progress);
+
             let mut splitstream = self.repo.create_stream(Some(*layer_sha256), None);
-            split_async(decoder, &mut splitstream).await?;
+            match descriptor.media_type() {
+                MediaType::ImageLayer => {
+                    split_async(progress, &mut splitstream).await?;
+                }
+                MediaType::ImageLayerGzip => {
+                    split_async(GzipDecoder::new(progress), &mut splitstream).await?;
+                }
+                MediaType::ImageLayerZstd => {
+                    split_async(ZstdDecoder::new(progress), &mut splitstream).await?;
+                }
+                other => bail!("Unsupported layer media type {:?}", other),
+            };
             let layer_id = self.repo.write_stream(splitstream, None)?;
             driver.await?;
             Ok(layer_id)