diff --git a/Makefile b/Makefile index 4852220..f6f5baf 100644 --- a/Makefile +++ b/Makefile @@ -57,7 +57,7 @@ format: @$(GO) fmt ./... test-unit: - @$(GO) test -v ./pkg/compose/... ./internal/... + @CGO_ENABLED=1 $(GO) test -race -v ./pkg/compose/... ./internal/... $(bd): @mkdir -p $@ diff --git a/cmd/composectl/cmd/fetch_progress.go b/cmd/composectl/cmd/fetch_progress.go new file mode 100644 index 0000000..6ed16fb --- /dev/null +++ b/cmd/composectl/cmd/fetch_progress.go @@ -0,0 +1,159 @@ +package composectl + +import ( + "fmt" + "io" + "sort" + "sync/atomic" + "time" + + "github.com/foundriesio/composeapp/pkg/compose" + "github.com/opencontainers/go-digest" +) + +// fetchProgressRenderer renders per-blob download progress. In a TTY it keeps +// only the in-flight blobs in an in-place redrawn region (bounded by the worker +// count) and "graduates" a blob to a single permanent line above that region +// once it completes. Because the live region never exceeds the worker count it +// cannot outgrow the terminal height, so the cursor-up redraw never clamps at +// the top of the screen (the defect that caused the same blob to be reprinted +// on several lines). Without a TTY each blob gets a line when it crosses each +// 10% of its size and a final completion line, so log collectors such as +// journalctl still depict download progress. +// +// Render is invoked serially from the progress reporter goroutine, so the +// renderer state needs no locking. +type fetchProgressRenderer struct { + out io.Writer + isTty bool + live []*compose.BlobFetchProgress // in-flight blobs currently redrawn, in stable order + seen map[digest.Digest]bool // blobs ever added to the live set + liveCount int // number of live lines drawn in the previous frame + reported map[digest.Digest]int64 // last reported progress decile per in-flight blob (non-TTY) +} + +func NewFetchProgressRenderer(out io.Writer, isTty bool) *fetchProgressRenderer { + return &fetchProgressRenderer{ + out: out, + isTty: isTty, + seen: map[digest.Digest]bool{}, + reported: map[digest.Digest]int64{}, + } +} + +// Render consumes one progress snapshot and updates the terminal accordingly. +func (r *fetchProgressRenderer) Render(p *compose.FetchProgress) { + done, live := r.track(p) + if r.isTty { + r.renderTty(done, live) + return + } + r.renderPlain(done, live) +} + +// track folds the latest snapshot into the live set: newly observed in-flight +// blobs are appended (deterministically ordered), and the live set is split into +// blobs that have just completed and those still in flight. r.live is left +// holding only the still-in-flight blobs. +func (r *fetchProgressRenderer) track(p *compose.FetchProgress) (done, live []*compose.BlobFetchProgress) { + var newcomers []*compose.BlobFetchProgress + for _, b := range p.Blobs { + if r.seen[b.Descriptor.Digest] { + continue + } + if b.State == compose.BlobFetching || b.State == compose.BlobOk { + r.seen[b.Descriptor.Digest] = true + newcomers = append(newcomers, b) + } + } + sort.Slice(newcomers, func(i, j int) bool { + if !newcomers[i].FetchStartTime.Equal(newcomers[j].FetchStartTime) { + return newcomers[i].FetchStartTime.Before(newcomers[j].FetchStartTime) + } + return newcomers[i].Descriptor.Digest < newcomers[j].Descriptor.Digest + }) + r.live = append(r.live, newcomers...) + + for _, b := range r.live { + if atomic.LoadInt64(&b.BytesFetched) >= b.Descriptor.Size { + done = append(done, b) + } else { + live = append(live, b) + } + } + r.live = live + return done, live +} + +// renderTty graduates completed blobs to permanent lines and redraws the live +// region in place. The leading cursor-up only covers the live region, so a +// terminal scroll of the permanent history above it is harmless. +func (r *fetchProgressRenderer) renderTty(done, live []*compose.BlobFetchProgress) { + if r.liveCount > 0 { + fmt.Fprintf(r.out, "\033[%dA", r.liveCount) + } + // Return to column 0 and clear from here to the end of the screen so stale + // live lines do not linger. + fmt.Fprint(r.out, "\r\033[J") + for _, b := range done { + fmt.Fprintf(r.out, " %s\n", formatBlobLine(b)) + } + for _, b := range live { + fmt.Fprintf(r.out, " %s\n", formatBlobLine(b)) + } + r.liveCount = len(live) +} + +// renderPlain handles non-TTY output (pipes, logs): cursor control is +// meaningless, so each in-flight blob gets a line when it first appears and +// then whenever it crosses another 10% of its size, plus one final line when +// it completes. +func (r *fetchProgressRenderer) renderPlain(done, live []*compose.BlobFetchProgress) { + for _, b := range done { + delete(r.reported, b.Descriptor.Digest) + fmt.Fprintf(r.out, " %s\n", formatBlobLine(b)) + } + for _, b := range live { + if r.crossedDecile(b) { + fmt.Fprintf(r.out, " %s\n", formatBlobLine(b)) + } + } +} + +// crossedDecile reports whether the blob has entered a 10%-of-size bucket that +// has not been reported yet, recording the new bucket if so. The first call for +// a blob always reports, so its download start shows up in the log. +func (r *fetchProgressRenderer) crossedDecile(b *compose.BlobFetchProgress) bool { + if b.Descriptor.Size <= 0 { + return false + } + decile := 100 * atomic.LoadInt64(&b.BytesFetched) / b.Descriptor.Size / 10 + last, ok := r.reported[b.Descriptor.Digest] + if ok && decile <= last { + return false + } + r.reported[b.Descriptor.Digest] = decile + return true +} + +// formatBlobLine renders a single blob's status line. All cross-goroutine +// counters are read via atomic loads since the fetch workers update them while +// this runs on the reporter goroutine. +func formatBlobLine(b *compose.BlobFetchProgress) string { + fetched := atomic.LoadInt64(&b.BytesFetched) + line := fmt.Sprintf("[%-12s] %.12s start: %s from: %10s ", + b.Type, + b.Descriptor.Digest.Encoded(), + b.FetchStartTime.UTC().Format(time.TimeOnly), + compose.FormatBytesInt64(b.BlobInfo.BytesFetched)) + line += fmt.Sprintf("progress: %10s / %10s (%3.0f%%) avg: %10s/s cur: %10s/s", + compose.FormatBytesInt64(fetched), + compose.FormatBytesInt64(b.Descriptor.Size), + 100*float64(fetched)/float64(b.Descriptor.Size), + compose.FormatBytesInt64(atomic.LoadInt64(&b.ReadSpeedAvg)), + compose.FormatBytesInt64(atomic.LoadInt64(&b.ReadSpeedCur))) + if fetched >= b.Descriptor.Size { + line += "; done at " + time.Now().UTC().Format(time.TimeOnly) + } + return line +} diff --git a/cmd/composectl/cmd/fetch_progress_test.go b/cmd/composectl/cmd/fetch_progress_test.go new file mode 100644 index 0000000..2d3f86c --- /dev/null +++ b/cmd/composectl/cmd/fetch_progress_test.go @@ -0,0 +1,187 @@ +package composectl + +import ( + "bytes" + "regexp" + "strconv" + "strings" + "testing" + "time" + + "github.com/foundriesio/composeapp/pkg/compose" + "github.com/opencontainers/go-digest" + ocispec "github.com/opencontainers/image-spec/specs-go/v1" +) + +var ( + ansiRe = regexp.MustCompile("\x1b\\[[0-9;]*[A-Za-z]") + cursorUpRe = regexp.MustCompile("\x1b\\[([0-9]+)A") +) + +// newProgressBlob builds a blob whose digest is unique per id, sized so progress +// can be advanced across several frames. +func newProgressBlob(t *testing.T, id string, size int64, start time.Time) *compose.BlobFetchProgress { + t.Helper() + dgst := digest.FromString(id) + return &compose.BlobFetchProgress{ + BlobInfo: compose.BlobInfo{ + Descriptor: &ocispec.Descriptor{Digest: dgst, Size: size}, + Type: compose.BlobTypeImageLayer, + State: compose.BlobFetching, + }, + FetchStartTime: start, + } +} + +func shortDigest(b *compose.BlobFetchProgress) string { + enc := b.Descriptor.Digest.Encoded() + return enc[:12] +} + +func snapshot(blobs ...*compose.BlobFetchProgress) *compose.FetchProgress { + m := compose.BlobsFetchProgress{} + for _, b := range blobs { + m[b.Descriptor.Digest] = b + } + return &compose.FetchProgress{Blobs: m} +} + +// advance sets a blob's fetched bytes and flips its state to done once complete. +func advance(b *compose.BlobFetchProgress, fetched int64) { + b.BytesFetched = fetched + if fetched >= b.Descriptor.Size { + b.State = compose.BlobOk + } +} + +func doneLineCount(output string, b *compose.BlobFetchProgress) int { + count := 0 + for _, line := range strings.Split(ansiRe.ReplaceAllString(output, ""), "\n") { + if strings.Contains(line, shortDigest(b)) && strings.Contains(line, "; done at") { + count++ + } + } + return count +} + +func anyLineFor(output string, b *compose.BlobFetchProgress) int { + count := 0 + for _, line := range strings.Split(ansiRe.ReplaceAllString(output, ""), "\n") { + if strings.Contains(line, shortDigest(b)) { + count++ + } + } + return count +} + +// runFrames feeds a fixed multi-frame scenario into the renderer: blob A finishes +// in frame 1, blob B spans all three frames, blob C appears in frame 2 and +// finishes in frame 3. +func runFrames(r *fetchProgressRenderer) (a, b, c *compose.BlobFetchProgress) { + t0 := time.Unix(0, 0).UTC() + a = &compose.BlobFetchProgress{ + BlobInfo: compose.BlobInfo{ + Descriptor: &ocispec.Descriptor{Digest: digest.FromString("blob-a"), Size: 10}, + Type: compose.BlobTypeImageManifest, + State: compose.BlobOk, + }, + FetchStartTime: t0, + BytesFetched: 10, + } + b = &compose.BlobFetchProgress{ + BlobInfo: compose.BlobInfo{ + Descriptor: &ocispec.Descriptor{Digest: digest.FromString("blob-b"), Size: 100}, + Type: compose.BlobTypeImageLayer, + State: compose.BlobFetching, + }, + FetchStartTime: t0.Add(time.Second), + BytesFetched: 30, + } + c = &compose.BlobFetchProgress{ + BlobInfo: compose.BlobInfo{ + Descriptor: &ocispec.Descriptor{Digest: digest.FromString("blob-c"), Size: 50}, + Type: compose.BlobTypeImageLayer, + State: compose.BlobFetching, + }, + FetchStartTime: t0.Add(2 * time.Second), + } + + r.Render(snapshot(a, b)) // frame 1: A done, B at 30% + advance(b, 70) // frame 2: B at 70%, C appears at 0% + r.Render(snapshot(a, b, c)) // + advance(b, 100) // frame 3: B and C both complete + advance(c, 50) // + r.Render(snapshot(a, b, c)) // + return a, b, c +} + +func TestFetchProgressRendererPlainReportsProgressAndCompletion(t *testing.T) { + var out bytes.Buffer + r := NewFetchProgressRenderer(&out, false) + a, b, c := runFrames(r) + + for _, blob := range []*compose.BlobFetchProgress{a, b, c} { + if got := doneLineCount(out.String(), blob); got != 1 { + t.Errorf("blob %s: expected exactly one completion line in non-TTY mode, got %d\n%s", + shortDigest(blob), got, out.String()) + } + } + // A completes within its first frame: just the completion line. + // B is seen at 30% and 70% before completing: two progress lines + done. + // C is seen at 0% before completing: one progress line + done. + for blob, want := range map[*compose.BlobFetchProgress]int{a: 1, b: 3, c: 2} { + if got := anyLineFor(out.String(), blob); got != want { + t.Errorf("blob %s: expected %d printed lines in non-TTY mode, got %d\n%s", + shortDigest(blob), want, got, out.String()) + } + } +} + +func TestFetchProgressRendererTtyGraduatesEachBlobOnce(t *testing.T) { + var out bytes.Buffer + r := NewFetchProgressRenderer(&out, true) + a, b, c := runFrames(r) + + for _, blob := range []*compose.BlobFetchProgress{a, b, c} { + if got := doneLineCount(out.String(), blob); got != 1 { + t.Errorf("blob %s: expected exactly one completion line in TTY mode, got %d\n%s", + shortDigest(blob), got, out.String()) + } + } + + // The live region (and thus every cursor-up move) must never exceed the peak + // number of concurrently in-flight blobs (here B and C in frame 2 => 2). + const maxConcurrent = 2 + for _, m := range cursorUpRe.FindAllStringSubmatch(out.String(), -1) { + n, err := strconv.Atoi(m[1]) + if err != nil { + t.Fatalf("failed to parse cursor-up count %q: %s", m[1], err) + } + if n > maxConcurrent { + t.Errorf("cursor moved up %d lines, exceeding the in-flight bound %d; "+ + "the redraw region is growing unbounded again", n, maxConcurrent) + } + } +} + +// Ensures the non-TTY decile reporting prints a line only when a blob enters a +// new 10% bucket, and never marks an in-flight blob as completed. +func TestFetchProgressRendererPlainReportsOncePerDecile(t *testing.T) { + var out bytes.Buffer + r := NewFetchProgressRenderer(&out, false) + b := newProgressBlob(t, "incomplete", 100, time.Unix(0, 0).UTC()) + + advance(b, 40) + r.Render(snapshot(b)) // first observation at 40% reports + advance(b, 49) + r.Render(snapshot(b)) // still in the 40% bucket: no new line + advance(b, 50) + r.Render(snapshot(b)) // crossed into the 50% bucket: reports + + if got := anyLineFor(out.String(), b); got != 2 { + t.Errorf("expected one line per crossed decile (2), got %d:\n%s", got, out.String()) + } + if got := doneLineCount(out.String(), b); got != 0 { + t.Errorf("expected no completion line for an in-flight blob, got %d:\n%s", got, out.String()) + } +} diff --git a/cmd/composectl/cmd/pull.go b/cmd/composectl/cmd/pull.go index 5df6b6b..5218f73 100644 --- a/cmd/composectl/cmd/pull.go +++ b/cmd/composectl/cmd/pull.go @@ -4,7 +4,6 @@ import ( "encoding/json" "fmt" "os" - "sync/atomic" "time" "github.com/containerd/containerd/platforms" @@ -20,9 +19,16 @@ type ( SrcStorePath string PrintUsageStat bool Quick bool + Workers uint } ) +const ( + MinWorkers = 1 + MaxWorkers = 10 + DefaultWorkers = 3 +) + var ( exitCodeInsufficientSpace int = 100 ) @@ -41,14 +47,24 @@ func init() { pullCmd.Flags().StringVarP(&opts.SrcStorePath, "source-store-path", "l", "", "A path to the source store root directory") pullCmd.Flags().BoolVarP(&opts.PrintUsageStat, "print-usage-stat", "p", false, "A flag to enable/disable usage statistic output to stderr") pullCmd.Flags().BoolVar(&opts.Quick, "quick", false, "Skip checking hash of app blobs; verify only their presence and size") + pullCmd.Flags().UintVarP(&opts.Workers, "workers", "w", DefaultWorkers, + fmt.Sprintf("Number of concurrent blob download workers in range %d-%d", MinWorkers, MaxWorkers)) pullCmd.Run = func(cmd *cobra.Command, args []string) { checkWatermark(opts.UsageWatermark) + checkWorkers(opts.Workers) pullApps(cmd, args, &opts) } rootCmd.AddCommand(pullCmd) } +func checkWorkers(workers uint) { + if workers < MinWorkers || workers > MaxWorkers { + DieNotNilWithCode(fmt.Errorf("invalid `--workers` value: %d; should be between %d and %d", + workers, MinWorkers, MaxWorkers), 1, "invalid argument") + } +} + func pullApps(cmd *cobra.Command, args []string, opts *pullOptions) { if len(args) > 1 { fmt.Printf("Pulling %d apps to %s\n", len(args), config.StoreRoot) @@ -78,7 +94,8 @@ func pullApps(cmd *cobra.Command, args []string, opts *pullOptions) { err := compose.FetchBlobs(cmd.Context(), config, cr.MissingBlobs, compose.WithProgressPollInterval(1000), compose.WithFetchProgress(getFetchProgressHandler()), - compose.WithSourcePath(opts.SrcStorePath)) + compose.WithSourcePath(opts.SrcStorePath), + compose.WithFetchWorkers(int(opts.Workers))) DieNotNil(err, "failed to fetch blobs") fmt.Println("\n\nApp blobs pull completed at " + time.Now().UTC().Format("15:04:05 02 Jan 2006")) } @@ -90,62 +107,6 @@ func pullApps(cmd *cobra.Command, args []string, opts *pullOptions) { } func getFetchProgressHandler() func(progress *compose.FetchProgress) { - var blobsBeingFetched []struct { - blob *compose.BlobFetchProgress - done bool - } isTty := term.IsTerminal(os.Stdout.Fd()) || os.Getenv("PARENT_HAS_TTY") == "1" - return func(p *compose.FetchProgress) { - currentBlobsBeingFetchedNumb := len(blobsBeingFetched) - for _, bi := range p.Blobs { - found := false - for _, bf := range blobsBeingFetched { - if bf.blob.Descriptor.Digest == bi.Descriptor.Digest { - found = true - break - } - } - if !found && (bi.State == compose.BlobFetching || bi.State == compose.BlobOk) { - blobsBeingFetched = append(blobsBeingFetched, struct { - blob *compose.BlobFetchProgress - done bool - }{blob: bi, done: false}) - } - } - - if isTty && currentBlobsBeingFetchedNumb > 0 { - fmt.Printf("\033[%dA\r", currentBlobsBeingFetchedNumb) - } - - for i := range blobsBeingFetched { - if blobsBeingFetched[i].done { - if isTty { - // Move cursor down one line since this blob fetch is completed - fmt.Print("\033[1B") - } - continue - } - - b := blobsBeingFetched[i].blob - - fmt.Printf("\n [%-12s] %.12s start: %s from: %10s ", - b.Type, - b.Descriptor.Digest.Encoded(), - b.FetchStartTime.UTC().Format(time.TimeOnly), - compose.FormatBytesInt64(b.BlobInfo.BytesFetched)) - - fmt.Printf("progress: %10s / %10s (%3.0f%%) avg: %10s/s cur: %10s/s", - compose.FormatBytesInt64(atomic.LoadInt64(&b.BytesFetched)), - compose.FormatBytesInt64(b.Descriptor.Size), - 100*float64(b.BytesFetched)/float64(b.Descriptor.Size), - compose.FormatBytesInt64(atomic.LoadInt64(&b.ReadSpeedAvg)), - compose.FormatBytesInt64(atomic.LoadInt64(&b.ReadSpeedCur))) - - if b.BytesFetched == b.Descriptor.Size { - fmt.Printf("; done at %s", - time.Now().UTC().Format(time.TimeOnly)) - blobsBeingFetched[i].done = true - } - } - } + return NewFetchProgressRenderer(os.Stdout, isTty).Render } diff --git a/go.mod b/go.mod index 5293d28..967fcc8 100644 --- a/go.mod +++ b/go.mod @@ -19,6 +19,7 @@ require ( github.com/spf13/cobra v1.8.0 go.etcd.io/bbolt v1.3.7 golang.org/x/net v0.26.0 + golang.org/x/sync v0.7.0 gopkg.in/yaml.v3 v3.0.1 ) @@ -71,7 +72,6 @@ require ( go.opentelemetry.io/otel/trace v1.21.0 // indirect golang.org/x/exp v0.0.0-20230713183714-613f0c0eb8a1 // indirect golang.org/x/mod v0.17.0 // indirect - golang.org/x/sync v0.7.0 // indirect golang.org/x/sys v0.29.0 // indirect golang.org/x/tools v0.21.1-0.20240508182429-e35e4ccd0d2d // indirect google.golang.org/genproto/googleapis/rpc v0.0.0-20230822172742-b8732ec3820d // indirect diff --git a/pkg/compose/fetch.go b/pkg/compose/fetch.go index e24369e..59cdc2c 100644 --- a/pkg/compose/fetch.go +++ b/pkg/compose/fetch.go @@ -9,6 +9,7 @@ import ( "github.com/containerd/containerd/errdefs" "github.com/foundriesio/composeapp/internal/progress" "github.com/opencontainers/go-digest" + "golang.org/x/sync/errgroup" "io" "sync" "sync/atomic" @@ -17,6 +18,7 @@ import ( const ( DefaultPollInterval = 300 // Default interval between polling/checking blob download status in milliseconds + DefaultFetchWorkers = 3 // Default number of concurrent blob-fetch workers ) type ( @@ -41,6 +43,7 @@ type ( ProgressHandler FetchProgressFunc ProgressPollInterval int // interval between polling/checking blob download status in milliseconds SourcePath string // path to the source directory containing blobs to fetch, if specified, the blobs will be fetched from this directory instead of remote registry + Workers int // number of concurrent blob-fetch workers; defaults to DefaultFetchWorkers when not positive } FetchOption func(*FetchOptions) @@ -79,6 +82,12 @@ func WithSourcePath(sourcePath string) FetchOption { } } +func WithFetchWorkers(workers int) FetchOption { + return func(opts *FetchOptions) { + opts.Workers = workers + } +} + func FetchBlobs(ctx context.Context, cfg *Config, blobs BlobsInfo, options ...FetchOption) error { opts := FetchOptions{} for _, o := range options { @@ -164,33 +173,11 @@ func FetchBlobs(ctx context.Context, cfg *Config, blobs BlobsInfo, options ...Fe }(stopChan) } - for _, bi := range getOrderedBlobsToFetch(blobsToFetch) { - err = func() error { - // Get the reader without digest calculation and verification because the writer/ingester of - // the local store (`ls`) will do that. - //r, err := blobProvider.GetReadCloser(ctx, WithRef(bi.Ref()), WithDescriptor(*bi.Descriptor)) - r, err := blobProvider.GetReadCloser(ctx, WithRef(bi.Ref()), WithDescriptor(*bi.Descriptor), WithSecureReadOff()) - if err != nil { - return fmt.Errorf("failed to initiate request to fetch blob %s: %w", bi.Descriptor.Digest, err) - } - defer r.Close() - blobReader, ok := r.(io.ReadSeekCloser) - if !ok { - return fmt.Errorf("blob fetch reader for %s does not implement io.ReadSeekCloser", bi.Ref()) - } - bi.FetchStartTime = time.Now() - rm := NewReadMonitor(ctx, blobReader, bi) - rm.Start() - defer rm.Stop() - if err := CopyBlob(ctx, rm, bi.Ref(), *bi.Descriptor, ls, true); err != nil { - return fmt.Errorf("failed to fetch blob %s: %w", bi.Descriptor.Digest, err) - } - return nil - }() - if err != nil { - break - } + workers := opts.Workers + if workers <= 0 { + workers = DefaultFetchWorkers } + err = runBlobFetchPool(ctx, blobProvider, ls, getOrderedBlobsToFetch(blobsToFetch), workers) if progressReporter != nil { if ctx.Err() == nil { @@ -205,6 +192,43 @@ func FetchBlobs(ctx context.Context, cfg *Config, blobs BlobsInfo, options ...Fe return ctx.Err() } +// runBlobFetchPool downloads the given blobs concurrently using a bounded pool of +// workers. It is fail-fast: the first error cancels the derived context so that +// any in-flight or queued downloads stop, and that error is returned. +func runBlobFetchPool(ctx context.Context, blobProvider BlobProvider, ls content.Store, blobs []*BlobFetchProgress, workers int) error { + g, gctx := errgroup.WithContext(ctx) + g.SetLimit(workers) + for _, bi := range blobs { + g.Go(func() error { + return fetchSingleBlob(gctx, blobProvider, ls, bi) + }) + } + return g.Wait() +} + +// fetchSingleBlob downloads one blob into the local store. The reader is created +// without digest calculation/verification because the local store ingester (`ls`) +// performs that during the copy. +func fetchSingleBlob(ctx context.Context, blobProvider BlobProvider, ls content.Store, bi *BlobFetchProgress) error { + r, err := blobProvider.GetReadCloser(ctx, WithRef(bi.Ref()), WithDescriptor(*bi.Descriptor), WithSecureReadOff()) + if err != nil { + return fmt.Errorf("failed to initiate request to fetch blob %s: %w", bi.Descriptor.Digest, err) + } + defer r.Close() + blobReader, ok := r.(io.ReadSeekCloser) + if !ok { + return fmt.Errorf("blob fetch reader for %s does not implement io.ReadSeekCloser", bi.Ref()) + } + bi.FetchStartTime = time.Now() + rm := NewReadMonitor(ctx, blobReader, bi) + rm.Start() + defer rm.Stop() + if err := CopyBlob(ctx, rm, bi.Ref(), *bi.Descriptor, ls, true); err != nil { + return fmt.Errorf("failed to fetch blob %s: %w", bi.Descriptor.Digest, err) + } + return nil +} + func checkAndUpdateBlobStatus(ctx context.Context, fetchProgress *FetchProgress, ls content.Store, sr progress.Reporter[FetchProgress]) { for _, b := range fetchProgress.Blobs { if b.State == BlobOk { diff --git a/pkg/compose/fetch_test.go b/pkg/compose/fetch_test.go new file mode 100644 index 0000000..3be6c7e --- /dev/null +++ b/pkg/compose/fetch_test.go @@ -0,0 +1,215 @@ +package compose + +import ( + "bytes" + "context" + "errors" + "fmt" + "io" + "sync" + "testing" + "time" + + "github.com/containerd/containerd/content" + "github.com/containerd/containerd/content/local" + "github.com/opencontainers/go-digest" + ocispec "github.com/opencontainers/image-spec/specs-go/v1" +) + +// errInjected is the sentinel a fakeBlobProvider wraps when asked to fail a blob, +// so tests can assert error propagation with errors.Is. +var errInjected = errors.New("injected fetch failure") + +// fakeReadSeekCloser serves blob bytes from memory and, on the first read, sleeps +// for `delay` to widen the window during which a fetch is "in flight" so the test +// can observe real overlap between concurrent workers. +type fakeReadSeekCloser struct { + *bytes.Reader + delay time.Duration + slept bool + onClose func() +} + +func (f *fakeReadSeekCloser) Read(p []byte) (int, error) { + if !f.slept { + f.slept = true + if f.delay > 0 { + time.Sleep(f.delay) + } + } + return f.Reader.Read(p) +} + +func (f *fakeReadSeekCloser) Close() error { + if f.onClose != nil { + f.onClose() + } + return nil +} + +// fakeBlobProvider is an in-memory BlobProvider that tracks how many fetches are +// concurrently in flight and can inject a failure for a specific digest. +type fakeBlobProvider struct { + blobs map[digest.Digest][]byte + failDigest digest.Digest + readDelay time.Duration + + mu sync.Mutex + active int + maxActive int +} + +func (p *fakeBlobProvider) Type() BlobProviderType { return BlobProviderTypeMemory } + +func (p *fakeBlobProvider) Info(_ context.Context, _ digest.Digest) (content.Info, error) { + return content.Info{}, fmt.Errorf("not implemented") +} + +func (p *fakeBlobProvider) GetReadCloser(_ context.Context, opts ...SecureReadOptions) (io.ReadCloser, error) { + dgst := GetSecureReadParams(opts...).Descriptor.Digest + if p.failDigest != "" && dgst == p.failDigest { + return nil, fmt.Errorf("blob %s: %w", dgst, errInjected) + } + data, ok := p.blobs[dgst] + if !ok { + return nil, fmt.Errorf("blob %s not found", dgst) + } + p.enter() + return &fakeReadSeekCloser{ + Reader: bytes.NewReader(data), + delay: p.readDelay, + onClose: p.leave, + }, nil +} + +func (p *fakeBlobProvider) enter() { + p.mu.Lock() + defer p.mu.Unlock() + p.active++ + if p.active > p.maxActive { + p.maxActive = p.active + } +} + +func (p *fakeBlobProvider) leave() { + p.mu.Lock() + defer p.mu.Unlock() + p.active-- +} + +func (p *fakeBlobProvider) maxConcurrency() int { + p.mu.Lock() + defer p.mu.Unlock() + return p.maxActive +} + +func newTestStore(t *testing.T) content.Store { + t.Helper() + ls, err := local.NewStore(t.TempDir()) + if err != nil { + t.Fatalf("failed to create local store: %s", err) + } + return ls +} + +// newTestBlob builds a fetchable blob descriptor whose digest matches its bytes so +// the local store's digest verification during CopyBlob succeeds. +func newTestBlob(data []byte) *BlobFetchProgress { + dgst := digest.FromBytes(data) + desc := &ocispec.Descriptor{ + MediaType: "application/octet-stream", + Digest: dgst, + Size: int64(len(data)), + URLs: []string{"fake://" + dgst.String()}, + } + return &BlobFetchProgress{ + BlobInfo: BlobInfo{ + Descriptor: desc, + State: BlobMissing, + Type: BlobTypeImageLayer, + }, + } +} + +func TestRunBlobFetchPoolRespectsWorkerLimit(t *testing.T) { + const ( + workers = 3 + numBlobs = 9 + ) + provider := &fakeBlobProvider{blobs: map[digest.Digest][]byte{}, readDelay: 100 * time.Millisecond} + var blobs []*BlobFetchProgress + for i := 0; i < numBlobs; i++ { + data := []byte(fmt.Sprintf("limit-blob-%d-payload", i)) + provider.blobs[digest.FromBytes(data)] = data + blobs = append(blobs, newTestBlob(data)) + } + + ls := newTestStore(t) + if err := runBlobFetchPool(context.Background(), provider, ls, blobs, workers); err != nil { + t.Fatalf("runBlobFetchPool returned an unexpected error: %s", err) + } + + // The pool must never run more than `workers` fetches at once, yet with more + // blobs than workers it must saturate the pool to exactly `workers`. + if got := provider.maxConcurrency(); got != workers { + t.Errorf("expected peak concurrency to equal the worker limit %d, got %d", workers, got) + } + + for _, b := range blobs { + if _, err := ls.Info(context.Background(), b.Descriptor.Digest); err != nil { + t.Errorf("blob %s missing from store after fetch: %s", b.Descriptor.Digest, err) + } + } +} + +func TestRunBlobFetchPoolFailsFast(t *testing.T) { + provider := &fakeBlobProvider{blobs: map[digest.Digest][]byte{}} + var blobs []*BlobFetchProgress + for i := 0; i < 5; i++ { + data := []byte(fmt.Sprintf("failfast-blob-%d", i)) + provider.blobs[digest.FromBytes(data)] = data + blobs = append(blobs, newTestBlob(data)) + } + // Inject a failure for one of the blobs. + provider.failDigest = blobs[2].Descriptor.Digest + + ls := newTestStore(t) + err := runBlobFetchPool(context.Background(), provider, ls, blobs, 2) + if err == nil { + t.Fatal("expected runBlobFetchPool to return an error, got nil") + } + if !errors.Is(err, errInjected) { + t.Errorf("expected returned error to wrap the injected failure, got: %s", err) + } + // The failing blob must never be committed to the store. + if _, infoErr := ls.Info(context.Background(), blobs[2].Descriptor.Digest); infoErr == nil { + t.Errorf("failing blob %s should not be present in the store", blobs[2].Descriptor.Digest) + } +} + +func TestRunBlobFetchPoolFetchesAllBlobs(t *testing.T) { + const workers = DefaultFetchWorkers + provider := &fakeBlobProvider{blobs: map[digest.Digest][]byte{}} + var blobs []*BlobFetchProgress + for i := 0; i < 5; i++ { + data := []byte(fmt.Sprintf("all-blob-%d-payload", i)) + provider.blobs[digest.FromBytes(data)] = data + blobs = append(blobs, newTestBlob(data)) + } + + ls := newTestStore(t) + if err := runBlobFetchPool(context.Background(), provider, ls, blobs, workers); err != nil { + t.Fatalf("runBlobFetchPool returned an unexpected error: %s", err) + } + + for _, b := range blobs { + info, err := ls.Info(context.Background(), b.Descriptor.Digest) + if err != nil { + t.Errorf("blob %s missing from store: %s", b.Descriptor.Digest, err) + continue + } + if info.Size != b.Descriptor.Size { + t.Errorf("blob %s stored size %d, want %d", b.Descriptor.Digest, info.Size, b.Descriptor.Size) + } + } +} diff --git a/test/compose/Dockerfile b/test/compose/Dockerfile index e5bd10b..1996fae 100644 --- a/test/compose/Dockerfile +++ b/test/compose/Dockerfile @@ -3,7 +3,7 @@ FROM golang:1.24-alpine ARG REG_CERT=test/compose/registry/certs/registry.crt ARG SRC_DIR=$PWD -RUN apk add make curl docker docker-compose git python3 py3-requests py3-expandvars py3-yaml skopeo golangci-lint +RUN apk add make curl docker docker-compose git python3 py3-requests py3-expandvars py3-yaml skopeo golangci-lint gcc musl-dev # fetch ci-scripts, needed for calculating sizes of extacted app layers RUN git clone https://github.com/foundriesio/ci-scripts.git /ci-scripts diff --git a/test/fixtures/composectl_cmds.go b/test/fixtures/composectl_cmds.go index c07472d..3a66b4b 100644 --- a/test/fixtures/composectl_cmds.go +++ b/test/fixtures/composectl_cmds.go @@ -17,6 +17,7 @@ import ( "os/exec" "path" "path/filepath" + "strconv" "testing" "time" ) @@ -192,6 +193,11 @@ func (a *App) Pull(t *testing.T) { a.runCmd(t, "pull app", "pull", a.PublishedUri, "-u", "90") } +func (a *App) PullWithWorkers(t *testing.T, workers int) { + t.Helper() + a.runCmd(t, "pull app", "pull", a.PublishedUri, "-u", "90", "-w", strconv.Itoa(workers)) +} + func (a *App) Remove(t *testing.T) { t.Helper() a.runCmd(t, "remove app", "rm", a.PublishedUri) @@ -444,6 +450,22 @@ func runCmd(t *testing.T, appDir string, args ...string) []byte { return output } +// RunCmdExpectFail runs composectl with the given args and fails the test unless +// the command exits with a non-zero status. It returns the combined output so the +// caller can assert on the user-facing error message. +func RunCmdExpectFail(t *testing.T, appDir string, args ...string) []byte { + t.Helper() + c := exec.Command(composeExec, args...) + if len(appDir) > 0 { + c.Dir = appDir + } + output, err := c.CombinedOutput() + if err == nil { + t.Fatalf("expected `%s` command to fail, but it succeeded; output: %s\n", args[0], output) + } + return output +} + func randomString(length int) string { const charset = "abcdefghijklmnopqrstuvwxyz0123456789" seededRand := rand2.New(rand2.NewSource(time.Now().UnixNano())) diff --git a/test/integration/concurrent_pull_test.go b/test/integration/concurrent_pull_test.go new file mode 100644 index 0000000..c522e6e --- /dev/null +++ b/test/integration/concurrent_pull_test.go @@ -0,0 +1,46 @@ +package e2e_tests + +import ( + "strings" + "testing" + + f "github.com/foundriesio/composeapp/test/fixtures" +) + +// TestPullWorkersFlagValidation asserts that out-of-range --workers values are +// rejected with a clear, user-facing error before any pull work begins. +func TestPullWorkersFlagValidation(t *testing.T) { + for _, tc := range []struct { + name string + value string + }{ + {"below minimum", "0"}, + {"above maximum", "11"}, + } { + t.Run(tc.name, func(t *testing.T) { + out := f.RunCmdExpectFail(t, "", + "pull", "registry:5000/factory/does-not-exist:latest", "-w", tc.value) + msg := string(out) + if !strings.Contains(msg, "invalid `--workers` value") || !strings.Contains(msg, "between 1 and 10") { + t.Fatalf("expected an --workers range validation error, got: %s", msg) + } + }) + } +} + +// TestPullWithConcurrentWorkers verifies the --workers flag plumbs through to the +// concurrent fetch and that an app pulled with multiple workers is fully fetched. +func TestPullWithConcurrentWorkers(t *testing.T) { + appComposeDef := ` +services: + busybox: + image: ghcr.io/foundriesio/busybox:1.36 + command: sh -c "while true; do sleep 60; done" +` + app := f.NewApp(t, appComposeDef) + app.Publish(t) + defer app.Remove(t) + + app.PullWithWorkers(t, 5) + app.CheckFetched(t) +}