allow for concurrent layer reads

Signed-off-by: Alex Goodman <[email protected]>

Author: wagoodman

go.mod

@@ -121,4 +121,6 @@ require (
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )
 
+require github.com/anchore/go-sync v0.0.0-20240306205607-3ee6b614d624
+
 replace github.com/gabriel-vasile/mimetype v1.4.4 => github.com/anchore/mimetype v0.0.0-20240710165720-f966690755a5

go.sum

@@ -23,6 +23,8 @@ github.com/anchore/go-collections v0.0.0-20240216171411-9321230ce537 h1:GjNGuwK5
 github.com/anchore/go-collections v0.0.0-20240216171411-9321230ce537/go.mod h1:1aiktV46ATCkuVg0O573ZrH56BUawTECPETbZyBcqT8=
 github.com/anchore/go-logger v0.0.0-20220728155337-03b66a5207d8 h1:imgMA0gN0TZx7PSa/pdWqXadBvrz8WsN6zySzCe4XX0=
 github.com/anchore/go-logger v0.0.0-20220728155337-03b66a5207d8/go.mod h1:+gPap4jha079qzRTUaehv+UZ6sSdaNwkH0D3b6zhTuk=
+github.com/anchore/go-sync v0.0.0-20240306205607-3ee6b614d624 h1:uKEb2vI/rlEGhgLs9BzRwCQDUVXyB8ia9vULpm/Q9Rc=
+github.com/anchore/go-sync v0.0.0-20240306205607-3ee6b614d624/go.mod h1:4gU9pKhRjTOiU34grx5w5IM7YxZfLx+9TrrultzHThU=
 github.com/anchore/go-testutils v0.0.0-20200925183923-d5f45b0d3c04 h1:VzprUTpc0vW0nnNKJfJieyH/TZ9UYAnTZs5/gHTdAe8=
 github.com/anchore/go-testutils v0.0.0-20200925183923-d5f45b0d3c04/go.mod h1:6dK64g27Qi1qGQZ67gFmBFvEHScy0/C8qhQhNe5B5pQ=
 github.com/anchore/mimetype v0.0.0-20240710165720-f966690755a5 h1:MGFRxHpfVyCoRRanIz6JJHZD2L4HLKifBGhGZReOLls=

pkg/image/docker/tarball_provider.go

@@ -38,7 +38,7 @@ func (p *tarballImageProvider) Name() string {
 }
 
 // Provide an image object that represents the docker image tar at the configured location on disk.
-func (p *tarballImageProvider) Provide(_ context.Context) (*image.Image, error) {
+func (p *tarballImageProvider) Provide(ctx context.Context) (*image.Image, error) {
 	img, err := tarball.ImageFromPath(p.path, nil)
 	if err != nil {
 		// raise a more controlled error for when there are multiple images within the given tar (from https://github.com/anchore/grype/issues/215)
@@ -92,7 +92,7 @@ func (p *tarballImageProvider) Provide(_ context.Context) (*image.Image, error)
 	}
 
 	out := image.New(img, p.tmpDirGen, contentTempDir, metadata...)
-	err = out.Read()
+	err = out.Read(ctx)
 	if err != nil {
 		return nil, err
 	}

pkg/image/image.go

@@ -1,6 +1,7 @@
 package image
 
 import (
+	"context"
 	"crypto/sha256"
 	"fmt"
 	"io"
@@ -14,6 +15,7 @@ import (
 	"github.com/wagoodman/go-partybus"
 	"github.com/wagoodman/go-progress"
 
+	"github.com/anchore/go-sync"
 	"github.com/anchore/stereoscope/internal/bus"
 	"github.com/anchore/stereoscope/internal/log"
 	"github.com/anchore/stereoscope/pkg/event"
@@ -192,10 +194,14 @@ func (i *Image) applyOverrideMetadata() error {
 	return nil
 }
 
+type layerReadResult struct {
+	layer *Layer
+	err   error
+}
+
 // Read parses information from the underlying image tar into this struct. This includes image metadata, layer
 // metadata, layer file trees, and layer squash trees (which implies the image squash tree).
-func (i *Image) Read() error {
-	var layers = make([]*Layer, 0)
+func (i *Image) Read(ctx context.Context) error {
 	var err error
 	i.Metadata, err = readImageMetadata(i.image)
 	if err != nil {
@@ -207,10 +213,7 @@ func (i *Image) Read() error {
 		return err
 	}
 
-	log.Debugf("image metadata: digest=%+v mediaType=%+v tags=%+v",
-		i.Metadata.ID,
-		i.Metadata.MediaType,
-		i.Metadata.Tags)
+	log.WithFields("digest", i.Metadata.ID, "mediaType", i.Metadata.MediaType, "tags", i.Metadata.Tags).Info("reading image")
 
 	v1Layers, err := i.image.Layers()
 	if err != nil {
@@ -222,19 +225,23 @@ func (i *Image) Read() error {
 
 	fileCatalog := NewFileCatalog()
 
-	for idx, v1Layer := range v1Layers {
-		layer := NewLayer(v1Layer)
-		err := layer.Read(fileCatalog, idx, i.contentCacheDir)
-		if err != nil {
-			return err
-		}
-		i.Metadata.Size += layer.Metadata.Size
-		layers = append(layers, layer)
+	exec, _ := sync.FromContext(ctx)
 
-		readProg.Increment()
+	collector := sync.NewCollector[layerReadResult](exec)
+
+	for idx, v1Layer := range v1Layers {
+		collector.Provide(readLayer(fileCatalog, idx, v1Layer, i.contentCacheDir, readProg))
 	}
 
-	i.Layers = layers
+	layerReadResults := collector.Collect()
+
+	for _, result := range layerReadResults {
+		if result.err != nil {
+			return result.err
+		}
+		i.Metadata.Size += result.layer.Metadata.Size
+		i.Layers = append(i.Layers, result.layer)
+	}
 
 	// in order to resolve symlinks all squashed trees must be available
 	err = i.squash(readProg)
@@ -245,6 +252,17 @@ func (i *Image) Read() error {
 	return err
 }
 
+func readLayer(fileCatalog *FileCatalog, idx int, v1Layer v1.Layer, contentDir string, prog *progress.Manual) sync.ProviderFunc[layerReadResult] {
+	return func() layerReadResult {
+		defer prog.Increment()
+		l := NewLayer(v1Layer)
+		return layerReadResult{
+			layer: l,
+			err:   l.Read(fileCatalog, idx, contentDir),
+		}
+	}
+}
+
 // squash generates a squash tree for each layer in the image. For instance, layer 2 squash =
 // squash(layer 0, layer 1, layer 2), layer 3 squash = squash(layer 0, layer 1, layer 2, layer 3), and so on.
 func (i *Image) squash(prog *progress.Manual) error {

pkg/image/layer.go

@@ -124,10 +124,7 @@ func (l *Layer) readStandardImageLayer(idx int, uncompressedLayersCacheDir strin
 		return err
 	}
 
-	log.Debugf("layer metadata: index=%+v digest=%+v mediaType=%+v",
-		l.Metadata.Index,
-		l.Metadata.Digest,
-		l.Metadata.MediaType)
+	log.WithFields("index", l.Metadata.Index, "digest", l.Metadata.Digest, "mediaType", l.Metadata.MediaType).Debug("reading layer")
 
 	tarFilePath, err := l.uncompressedCache(uncompressedLayersCacheDir)
 	if err != nil {
@@ -153,10 +150,7 @@ func (l *Layer) readSingularityImageLayer(idx int, uncompressedLayersCacheDir st
 		return err
 	}
 
-	log.Debugf("layer metadata: index=%+v digest=%+v mediaType=%+v",
-		l.Metadata.Index,
-		l.Metadata.Digest,
-		l.Metadata.MediaType)
+	log.WithFields("index", l.Metadata.Index, "digest", l.Metadata.Digest, "mediaType", l.Metadata.MediaType).Debug("reading layer")
 
 	monitor := trackReadProgress(l.Metadata)
 	sqfsFilePath, err := l.uncompressedCache(uncompressedLayersCacheDir)

pkg/image/oci/directory_provider.go

@@ -32,7 +32,7 @@ func (p *directoryImageProvider) Name() string {
 }
 
 // Provide an image object that represents the OCI image as a directory.
-func (p *directoryImageProvider) Provide(_ context.Context) (*image.Image, error) {
+func (p *directoryImageProvider) Provide(ctx context.Context) (*image.Image, error) {
 	pathObj, err := layout.FromPath(p.path)
 	if err != nil {
 		return nil, fmt.Errorf("unable to read image from OCI directory path %q: %w", p.path, err)
@@ -81,7 +81,7 @@ func (p *directoryImageProvider) Provide(_ context.Context) (*image.Image, error
 	}
 
 	out := image.New(img, p.tmpDirGen, contentTempDir, metadata...)
-	err = out.Read()
+	err = out.Read(ctx)
 	if err != nil {
 		return nil, err
 	}

pkg/image/oci/registry_provider.go

@@ -90,7 +90,7 @@ func (p *registryImageProvider) Provide(ctx context.Context) (*image.Image, erro
 	}
 
 	out := image.New(img, p.tmpDirGen, imageTempDir, metadata...)
-	err = out.Read()
+	err = out.Read(ctx)
 	if err != nil {
 		return nil, err
 	}

pkg/image/sif/archive_provider.go

@@ -31,7 +31,7 @@ func (p *singularityImageProvider) Name() string {
 }
 
 // Provide returns an Image that represents a Singularity Image Format (SIF) image.
-func (p *singularityImageProvider) Provide(_ context.Context) (*image.Image, error) {
+func (p *singularityImageProvider) Provide(ctx context.Context) (*image.Image, error) {
 	// We need to map the SIF to a GGCR v1.Image. Start with an implementation of the GGCR
 	// partial.UncompressedImageCore interface.
 	si, err := newSIFImage(p.path)
@@ -58,7 +58,7 @@ func (p *singularityImageProvider) Provide(_ context.Context) (*image.Image, err
 	}
 
 	out := image.New(ui, p.tmpDirGen, contentCacheDir, metadata...)
-	err = out.Read()
+	err = out.Read(ctx)
 	if err != nil {
 		return nil, err
 	}

pkg/image/sif/archive_provider_test.go

@@ -43,7 +43,7 @@ func TestSingularityImageProvider_Provide(t *testing.T) {
 			}
 
 			if err == nil {
-				if err := i.Read(); err != nil {
+				if err := i.Read(context.Background()); err != nil {
 					t.Fatal(err)
 				}
 			}

test/integration/oci_registry_source_test.go

@@ -38,7 +38,7 @@ func TestOciRegistrySourceMetadata(t *testing.T) {
 		require.NoError(t, img.Cleanup())
 	})
 
-	require.NoError(t, img.Read())
+	require.NoError(t, img.Read(context.Background()))
 
 	assert.Len(t, img.Metadata.RepoDigests, 1)
 	assert.Equal(t, "index.docker.io/"+ref, img.Metadata.RepoDigests[0])