fixup! image/copy: Fix missing progress reporting for chunked layers

Signed-off-by: Marek Simek <msimek@redhat.com>

Author: simek-m

image/copy/blob.go

@@ -19,6 +19,7 @@ import (
 func (ic *imageCopier) copyBlobFromStream(ctx context.Context, srcReader io.Reader, srcInfo types.BlobInfo,
 	getOriginalLayerCopyWriter func(decompressor compressiontypes.DecompressorFunc) io.Writer,
 	isConfig bool, toEncrypt bool, bar *progressBar, layerIndex int, emptyLayer bool,
+	reporter *progressReporter,
 ) (types.BlobInfo, error) {
 	// The copying happens through a pipeline of connected io.Readers;
 	// that pipeline is built by updating stream.
@@ -84,16 +85,9 @@ func (ic *imageCopier) copyBlobFromStream(ctx context.Context, srcReader io.Read
 		return types.BlobInfo{}, err
 	}
 
-	// === Report progress using the ic.c.options.Progress channel, if required.
-	if ic.c.options.Progress != nil && ic.c.options.ProgressInterval > 0 {
-		progressReader := newProgressReader(
-			stream.reader,
-			ic.c.options.Progress,
-			ic.c.options.ProgressInterval,
-			srcInfo,
-		)
-		defer progressReader.reportDone()
-		stream.reader = progressReader
+	// === Wrap stream with progress reporting if a reporter was provided.
+	if reporter != nil {
+		stream.reader = newProgressReader(stream.reader, reporter)
 	}
 
 	// === Finally, send the layer stream to dest.

image/copy/blob_chunk_accessor_proxy.go

@@ -1,11 +1,15 @@
 package copy
 
 import (
+	"context"
 	"fmt"
 	"io"
+	"math"
+	"time"
 
 	"github.com/vbauerster/mpb/v8"
 	"github.com/vbauerster/mpb/v8/decor"
+	"go.podman.io/image/v5/internal/private"
 	"go.podman.io/image/v5/types"
 )
 
@@ -140,3 +144,38 @@ func (bar *progressBar) mark100PercentComplete() {
 		bar.SetTotal(-1, true) // total < 0 = set it to bar.Current(), report it; and mark the bar as complete.
 	}
 }
+
+// blobChunkAccessorProxy wraps a BlobChunkAccessor to update a *progressBar
+// and optionally *progressReporter (if non-nil) with the number of received bytes.
+type blobChunkAccessorProxy struct {
+	wrapped  private.BlobChunkAccessor // The underlying BlobChunkAccessor
+	bar      *progressBar              // A progress bar updated with the number of bytes read so far
+	reporter *progressReporter         // A progress reporter updated with the number of bytes read so far
+}
+
+// GetBlobAt returns a sequential channel of readers that contain data for the requested
+// blob chunks, and a channel that might get a single error value.
+// The specified chunks must be not overlapping and sorted by their offset.
+// The readers must be fully consumed, in the order they are returned, before blocking
+// to read the next chunk.
+// If the Length for the last chunk is set to math.MaxUint64, then it
+// fully fetches the remaining data from the offset to the end of the blob.
+func (s *blobChunkAccessorProxy) GetBlobAt(ctx context.Context, info types.BlobInfo, chunks []private.ImageSourceChunk) (chan io.ReadCloser, chan error, error) {
+	start := time.Now()
+	rc, errs, err := s.wrapped.GetBlobAt(ctx, info, chunks)
+	if err == nil {
+		total := int64(0)
+		for _, c := range chunks {
+			// do not update the progress bar if there is a chunk with unknown length.
+			if c.Length == math.MaxUint64 {
+				return rc, errs, err
+			}
+			total += int64(c.Length)
+		}
+		if s.reporter != nil {
+			s.reporter.reportRead(uint64(total))
+		}
+		s.bar.EwmaIncrInt64(total, time.Since(start))
+	}
+	return rc, errs, err
+}

image/copy/progress_bars.go image/copy/progress.goimage/copy/progress_bars.go renamed to image/copy/progress.go

@@ -15,20 +15,47 @@ type progressReporter struct {
 	artifact     types.BlobInfo                  // The blob metadata which is currently being progressed
 	lastUpdate   time.Time                       // The last time a progress channel event was sent
 	offset       uint64                          // The currently downloaded size in bytes
-	offsetUpdate uint64                          // The number of bytes downloaded within the last update interval
+	offsetUpdate uint64                          // The number of bytes downloaded since lastUpdate
 }
 
-// reportNewArtifact fires types.ProgressEventNewArtifact to its progress channel.
-func (r *progressReporter) reportNewArtifact() {
-	r.channel <- types.ProgressProperties{
+// newProgressReporter creates a new progress reporter
+// and immediately reports a new artifact event.
+func newProgressReporter(
+	channel chan<- types.ProgressProperties,
+	interval time.Duration,
+	artifact types.BlobInfo,
+) *progressReporter {
+	channel <- types.ProgressProperties{
 		Event:    types.ProgressEventNewArtifact,
-		Artifact: r.artifact,
+		Artifact: artifact,
+	}
+	return &progressReporter{
+		channel:    channel,
+		interval:   interval,
+		artifact:   artifact,
+		lastUpdate: time.Now(),
+	}
+}
+
+// reset resets the reporters progress
+// and reports its zeroed state.
+// It's meant to be used on error when
+// the processing has to be re-started
+// (e.g. ErrFallbackToOrdinaryLayerDownload).
+func (r *progressReporter) reset() {
+	r.offset = 0
+	r.offsetUpdate = 0
+
+	r.channel <- types.ProgressProperties{
+		Event:        types.ProgressEventRead,
+		Artifact:     r.artifact,
+		Offset:       r.offset,
+		OffsetUpdate: r.offsetUpdate,
 	}
 	r.lastUpdate = time.Now()
 }
 
-// reportRead fires the types.ProgressEventRead event with `bytesRead`
-// to its progress channel.
+// reportRead reports progress with the number of `bytesRead`.
 func (r *progressReporter) reportRead(bytesRead uint64) {
 	r.offset += bytesRead
 	r.offsetUpdate += bytesRead
@@ -44,7 +71,7 @@ func (r *progressReporter) reportRead(bytesRead uint64) {
 	}
 }
 
-// reportDone fires the ProgressEventDone to its progress channel.
+// reportDone reports completion.
 func (r *progressReporter) reportDone() {
 	r.channel <- types.ProgressProperties{
 		Event:        types.ProgressEventDone,
@@ -54,34 +81,23 @@ func (r *progressReporter) reportDone() {
 	}
 }
 
-// progressReader is an io.Reader that reports its progress to
-// an underlying *progressReporter.
+// progressReader extends a wrapped io.Reader
+// with additional reporting of its progress.
 type progressReader struct {
 	source io.Reader
 	*progressReporter
 }
 
-// newProgressReader creates a new progress reader for
-// `source`:   The source when internally reading bytes
-// `channel`:  The reporter channel to which the progress will be sent
-// `interval`: The update interval to indicate how often the progress should update
-// `artifact`: The blob metadata which is currently being progressed.
+// newProgressReader creates a new progress reader that wraps source
+// and reports progress through the given reporter.
 func newProgressReader(
 	source io.Reader,
-	channel chan<- types.ProgressProperties,
-	interval time.Duration,
-	artifact types.BlobInfo,
+	reporter *progressReporter,
 ) *progressReader {
-	r := &progressReader{
-		source: source,
-		progressReporter: &progressReporter{
-			channel:  channel,
-			interval: interval,
-			artifact: artifact,
-		},
+	return &progressReader{
+		source:           source,
+		progressReporter: reporter,
 	}
-	r.reportNewArtifact()
-	return r
 }
 
 // Read continuously reads bytes into the progress reader and reports the

image/copy/progress_bars_test.go

@@ -18,19 +18,17 @@ func newSUT(
 ) *progressReader {
 	artifact := types.BlobInfo{Size: 10}
 
-	go func() {
-		res := <-channel
-		assert.Equal(t, res.Event, types.ProgressEventNewArtifact)
-		assert.Equal(t, res.Artifact, artifact)
-	}()
-	res := newProgressReader(reader, channel, duration, artifact)
+	reporter := newProgressReporter(channel, duration, artifact)
+	res := <-channel
+	assert.Equal(t, res.Event, types.ProgressEventNewArtifact)
+	assert.Equal(t, res.Artifact, artifact)
 
-	return res
+	return newProgressReader(reader, reporter)
 }
 
 func TestNewProgressReader(t *testing.T) {
 	// Given
-	channel := make(chan types.ProgressProperties)
+	channel := make(chan types.ProgressProperties, 1)
 	sut := newSUT(t, nil, time.Second, channel)
 	assert.NotNil(t, sut)
 
@@ -44,7 +42,7 @@ func TestNewProgressReader(t *testing.T) {
 
 func TestReadWithoutEvent(t *testing.T) {
 	// Given
-	channel := make(chan types.ProgressProperties)
+	channel := make(chan types.ProgressProperties, 1)
 	reader := bytes.NewReader([]byte{0, 1, 2})
 	sut := newSUT(t, reader, time.Second, channel)
 	assert.NotNil(t, sut)
@@ -60,7 +58,7 @@ func TestReadWithoutEvent(t *testing.T) {
 
 func TestReadWithEvent(t *testing.T) {
 	// Given
-	channel := make(chan types.ProgressProperties)
+	channel := make(chan types.ProgressProperties, 1)
 	reader := bytes.NewReader([]byte{0, 1, 2, 3, 4, 5, 6})
 	sut := newSUT(t, reader, time.Nanosecond, channel)
 	assert.NotNil(t, sut)