From 8e6a547d24dda39df1a5ed0e36c7744d48f8788e Mon Sep 17 00:00:00 2001 From: Gautam Raj <156491596+gautam-rl@users.noreply.github.com> Date: Fri, 8 Aug 2025 13:28:02 -0700 Subject: [PATCH 1/2] Retry on docker registry upload failures --- cmd/convertor/builder/builder.go | 6 +- cmd/convertor/builder/builder_engine.go | 5 +- cmd/convertor/builder/builder_utils.go | 130 ++++++++++++++++----- cmd/convertor/builder/overlaybd_builder.go | 8 +- cmd/convertor/builder/turboOCI_builder.go | 6 +- cmd/convertor/main.go | 4 + 6 files changed, 122 insertions(+), 37 deletions(-) diff --git a/cmd/convertor/builder/builder.go b/cmd/convertor/builder/builder.go index b3776c6c..c32e053b 100644 --- a/cmd/convertor/builder/builder.go +++ b/cmd/convertor/builder/builder.go @@ -75,6 +75,9 @@ type BuilderOptions struct { // Push manifests with subject Referrer bool + // Number of retries for registry upload operations when encountering 429 rate limiting + RetryCount int + // CustomResolver allows using a custom resolver instead of the default docker resolver // Used for tar import/export functionality CustomResolver remotes.Resolver @@ -223,7 +226,7 @@ func (b *graphBuilder) process(ctx context.Context, src v1.Descriptor, tag bool) } else { pusher = b.pusher } - if err := uploadBytes(ctx, pusher, expected, indexBytes); err != nil { + if err := uploadBytesWithRetry(ctx, pusher, expected, indexBytes, b.RetryCount); err != nil { return v1.Descriptor{}, fmt.Errorf("failed to upload index: %w", err) } log.G(ctx).Infof("index uploaded, %s", expected.Digest) @@ -296,6 +299,7 @@ func (b *graphBuilder) buildOne(ctx context.Context, src v1.Descriptor, tag bool engineBase.reserve = b.Reserve engineBase.noUpload = b.NoUpload engineBase.dumpManifest = b.DumpManifest + engineBase.retryCount = b.RetryCount if _, ok := b.Resolver.(*FileBasedResolver); ok { engineBase.tarExport = true } diff --git a/cmd/convertor/builder/builder_engine.go b/cmd/convertor/builder/builder_engine.go index c37cd28a..bccb6efa 100644 --- a/cmd/convertor/builder/builder_engine.go +++ b/cmd/convertor/builder/builder_engine.go @@ -117,6 +117,7 @@ type builderEngineBase struct { dumpManifest bool referrer bool tarExport bool + retryCount int } func (e *builderEngineBase) isGzipLayer(ctx context.Context, idx int) (bool, error) { @@ -183,7 +184,7 @@ func (e *builderEngineBase) uploadManifestAndConfig(ctx context.Context) (specs. Size: (int64)(len(cbuf)), } if shouldUploadBlob { - if err = uploadBytes(ctx, e.pusher, e.manifest.Config, cbuf); err != nil { + if err = uploadBytesWithRetry(ctx, e.pusher, e.manifest.Config, cbuf, e.retryCount); err != nil { return specs.Descriptor{}, errors.Wrapf(err, "failed to upload config") } log.G(ctx).Infof("config uploaded") @@ -207,7 +208,7 @@ func (e *builderEngineBase) uploadManifestAndConfig(ctx context.Context) (specs. Size: (int64)(len(cbuf)), } if shouldUploadBlob { - if err = uploadBytes(ctx, e.pusher, manifestDesc, cbuf); err != nil { + if err = uploadBytesWithRetry(ctx, e.pusher, manifestDesc, cbuf, e.retryCount); err != nil { return specs.Descriptor{}, errors.Wrapf(err, "failed to upload manifest") } e.outputDesc = manifestDesc diff --git a/cmd/convertor/builder/builder_utils.go b/cmd/convertor/builder/builder_utils.go index 6fad2c40..d6c83189 100644 --- a/cmd/convertor/builder/builder_utils.go +++ b/cmd/convertor/builder/builder_utils.go @@ -24,12 +24,15 @@ import ( "encoding/json" "fmt" "io" + "math" "os" "path" + "time" "github.com/containerd/containerd/v2/core/content" "github.com/containerd/containerd/v2/core/images" "github.com/containerd/containerd/v2/core/remotes" + "github.com/containerd/containerd/v2/core/remotes/docker" "github.com/containerd/containerd/v2/pkg/archive/compression" "github.com/containerd/continuity" "github.com/containerd/errdefs" @@ -43,6 +46,63 @@ import ( t "github.com/containerd/accelerated-container-image/pkg/types" ) +// isRetryableError checks if the error is retryable (429 or 5xx errors) +func isRetryableError(err error) bool { + if err == nil { + return false + } + + // Check for containerd docker error types + var dockerErr *docker.Error + if errors.As(err, &dockerErr) { + switch dockerErr.Code { + case docker.ErrorCodeTooManyRequests: + return true + case docker.ErrorCodeUnavailable: + return true + default: + return false + } + } + + return false +} + +// retryWithBackoff executes a function with exponential backoff on retryable errors +func retryWithBackoff(ctx context.Context, maxRetries int, operation func() error) error { + var lastErr error + + for attempt := 0; attempt <= maxRetries; attempt++ { + lastErr = operation() + + if lastErr == nil { + return nil + } + + if !isRetryableError(lastErr) { + return lastErr + } + + if attempt == maxRetries { + logrus.Warnf("max retries (%d) reached for retryable error: %v", maxRetries, lastErr) + return lastErr + } + + // Exponential backoff: base delay of 1s, max 30s + backoffDelay := time.Duration(math.Min(float64(time.Second)*math.Pow(2, float64(attempt)), float64(30*time.Second))) + logrus.Infof("received retryable error, retrying in %v (attempt %d/%d): %v", backoffDelay, attempt+1, maxRetries, lastErr) + + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(backoffDelay): + continue + } + } + + return lastErr +} + func fetch(ctx context.Context, fetcher remotes.Fetcher, desc specs.Descriptor, target any) error { rc, err := fetcher.Fetch(ctx, desc) if err != nil { @@ -200,41 +260,57 @@ func getFileDesc(filepath string, decompress bool) (specs.Descriptor, error) { } func uploadBlob(ctx context.Context, pusher remotes.Pusher, path string, desc specs.Descriptor) error { - cw, err := pusher.Push(ctx, desc) - if err != nil { - if errdefs.IsAlreadyExists(err) { - logrus.Infof("layer %s exists", desc.Digest.String()) - return nil + return uploadBlobWithRetry(ctx, pusher, path, desc, 0) +} + +func uploadBlobWithRetry(ctx context.Context, pusher remotes.Pusher, path string, desc specs.Descriptor, retryCount int) error { + return retryWithBackoff(ctx, retryCount, func() error { + cw, err := pusher.Push(ctx, desc) + if err != nil { + if errdefs.IsAlreadyExists(err) { + logrus.Infof("layer %s exists", desc.Digest.String()) + return nil + } + return err } - return err - } - defer cw.Close() - fobd, err := os.Open(path) - if err != nil { - return err - } - defer fobd.Close() - if err = content.Copy(ctx, cw, fobd, desc.Size, desc.Digest); err != nil { - return err - } - return nil + defer cw.Close() + fobd, err := os.Open(path) + if err != nil { + return err + } + defer fobd.Close() + if err = content.Copy(ctx, cw, fobd, desc.Size, desc.Digest); err != nil { + return err + } + return nil + }) } func uploadBytes(ctx context.Context, pusher remotes.Pusher, desc specs.Descriptor, data []byte) error { - cw, err := pusher.Push(ctx, desc) - if err != nil { - if errdefs.IsAlreadyExists(err) { - logrus.Infof("content %s exists", desc.Digest.String()) - return nil + return uploadBytesWithRetry(ctx, pusher, desc, data, 0) +} + +func uploadBytesWithRetry(ctx context.Context, pusher remotes.Pusher, desc specs.Descriptor, data []byte, retryCount int) error { + return retryWithBackoff(ctx, retryCount, func() error { + cw, err := pusher.Push(ctx, desc) + if err != nil { + if errdefs.IsAlreadyExists(err) { + logrus.Infof("content %s exists", desc.Digest.String()) + return nil + } + return err } - return err - } - defer cw.Close() - return content.Copy(ctx, cw, bytes.NewReader(data), desc.Size, desc.Digest) + defer cw.Close() + return content.Copy(ctx, cw, bytes.NewReader(data), desc.Size, desc.Digest) + }) } func tagPreviouslyConvertedManifest(ctx context.Context, pusher remotes.Pusher, fetcher remotes.Fetcher, desc specs.Descriptor) error { + return tagPreviouslyConvertedManifestWithRetry(ctx, pusher, fetcher, desc, 0) +} + +func tagPreviouslyConvertedManifestWithRetry(ctx context.Context, pusher remotes.Pusher, fetcher remotes.Fetcher, desc specs.Descriptor, retryCount int) error { manifest := specs.Manifest{} if err := fetch(ctx, fetcher, desc, &manifest); err != nil { return fmt.Errorf("failed to fetch converted manifest: %w", err) @@ -243,7 +319,7 @@ func tagPreviouslyConvertedManifest(ctx context.Context, pusher remotes.Pusher, if err != nil { return err } - if err := uploadBytes(ctx, pusher, desc, cbuf); err != nil { + if err := uploadBytesWithRetry(ctx, pusher, desc, cbuf, retryCount); err != nil { return fmt.Errorf("failed to tag converted manifest: %w", err) } return nil diff --git a/cmd/convertor/builder/overlaybd_builder.go b/cmd/convertor/builder/overlaybd_builder.go index 99d52623..b5a757ec 100644 --- a/cmd/convertor/builder/overlaybd_builder.go +++ b/cmd/convertor/builder/overlaybd_builder.go @@ -192,7 +192,7 @@ func (e *overlaybdBuilderEngine) UploadLayer(ctx context.Context, idx int) error } shouldUploadBlob := !e.noUpload || e.tarExport if shouldUploadBlob { - if err := uploadBlob(ctx, e.pusher, path.Join(layerDir, commitFile), desc); err != nil { + if err := uploadBlobWithRetry(ctx, e.pusher, path.Join(layerDir, commitFile), desc, e.retryCount); err != nil { return errors.Wrapf(err, "failed to upload layer %d", idx) } } @@ -354,7 +354,7 @@ func (e *overlaybdBuilderEngine) CheckForConvertedManifest(ctx context.Context) // If a converted manifest has been found we still need to tag it to match the expected output tag. func (e *overlaybdBuilderEngine) TagPreviouslyConvertedManifest(ctx context.Context, desc specs.Descriptor) error { - return tagPreviouslyConvertedManifest(ctx, e.pusher, e.fetcher, desc) + return tagPreviouslyConvertedManifestWithRetry(ctx, e.pusher, e.fetcher, desc, e.retryCount) } // mountImage is responsible for mounting a specific manifest from a source repository, this includes @@ -391,7 +391,7 @@ func (e *overlaybdBuilderEngine) mountImage(ctx context.Context, manifest specs. if err != nil { return err } - return uploadBytes(ctx, e.pusher, desc, cbuf) + return uploadBytesWithRetry(ctx, e.pusher, desc, cbuf, e.retryCount) } func (e *overlaybdBuilderEngine) StoreConvertedManifestDetails(ctx context.Context) error { @@ -483,7 +483,7 @@ func (e *overlaybdBuilderEngine) uploadBaseLayer(ctx context.Context) (specs.Des } shouldUploadBlob := !e.noUpload || e.tarExport if shouldUploadBlob { - if err = uploadBlob(ctx, e.pusher, tarFile, baseDesc); err != nil { + if err = uploadBlobWithRetry(ctx, e.pusher, tarFile, baseDesc, e.retryCount); err != nil { return specs.Descriptor{}, errors.Wrapf(err, "failed to upload baselayer") } logrus.Infof("baselayer uploaded") diff --git a/cmd/convertor/builder/turboOCI_builder.go b/cmd/convertor/builder/turboOCI_builder.go index a7a67e92..4a8c175f 100644 --- a/cmd/convertor/builder/turboOCI_builder.go +++ b/cmd/convertor/builder/turboOCI_builder.go @@ -168,7 +168,7 @@ func (e *turboOCIBuilderEngine) UploadLayer(ctx context.Context, idx int) error } } desc.Annotations[label.TurboOCIMediaType] = targetMediaType - if err := uploadBlob(ctx, e.pusher, path.Join(layerDir, tociLayerTar), desc); err != nil { + if err := uploadBlobWithRetry(ctx, e.pusher, path.Join(layerDir, tociLayerTar), desc, e.retryCount); err != nil { return errors.Wrapf(err, "failed to upload layer %d", idx) } e.tociLayers[idx] = desc @@ -196,7 +196,7 @@ func (e *turboOCIBuilderEngine) UploadImage(ctx context.Context) (specs.Descript }, } if !e.mkfs { - if err := uploadBlob(ctx, e.pusher, overlaybdBaseLayer, baseDesc); err != nil { + if err := uploadBlobWithRetry(ctx, e.pusher, overlaybdBaseLayer, baseDesc, e.retryCount); err != nil { return specs.Descriptor{}, errors.Wrapf(err, "failed to upload baselayer %q", overlaybdBaseLayer) } e.manifest.Layers = append([]specs.Descriptor{baseDesc}, e.manifest.Layers...) @@ -215,7 +215,7 @@ func (e *turboOCIBuilderEngine) UploadImage(ctx context.Context) (specs.Descript // If a converted manifest has been found we still need to tag it to match the expected output tag. func (e *turboOCIBuilderEngine) TagPreviouslyConvertedManifest(ctx context.Context, desc specs.Descriptor) error { - return tagPreviouslyConvertedManifest(ctx, e.pusher, e.fetcher, desc) + return tagPreviouslyConvertedManifestWithRetry(ctx, e.pusher, e.fetcher, desc, e.retryCount) } // Layer deduplication in FastOCI is not currently supported due to conversion not diff --git a/cmd/convertor/main.go b/cmd/convertor/main.go index c3342173..023e8ce5 100644 --- a/cmd/convertor/main.go +++ b/cmd/convertor/main.go @@ -59,6 +59,7 @@ var ( concurrencyLimit int disableSparse bool referrer bool + retryCount int // tar import/export importTar string @@ -279,6 +280,7 @@ Version: ` + commitID, ConcurrencyLimit: concurrencyLimit, DisableSparse: disableSparse, Referrer: referrer, + RetryCount: retryCount, } } else { // Normal registry mode @@ -307,6 +309,7 @@ Version: ` + commitID, ConcurrencyLimit: concurrencyLimit, DisableSparse: disableSparse, Referrer: referrer, + RetryCount: retryCount, } } if overlaybd != "" { @@ -393,6 +396,7 @@ func init() { rootCmd.Flags().IntVar(&concurrencyLimit, "concurrency-limit", 4, "the number of manifests that can be built at the same time, used for multi-arch images, 0 means no limit") rootCmd.Flags().BoolVar(&disableSparse, "disable-sparse", false, "disable sparse file for overlaybd") rootCmd.Flags().BoolVar(&referrer, "referrer", false, "push converted manifests with subject, note '--oci' will be enabled automatically if '--referrer' is set, cause the referrer must be in OCI format.") + rootCmd.Flags().IntVar(&retryCount, "retry-count", 5, "number of retries for registry upload operations when encountering 429 rate limiting") // tar import/export rootCmd.Flags().StringVar(&importTar, "import-tar", "", "import image from tar file (OCI layout format)") From 5c2bc21ffbfeb5aa8922ef80f8628c17995d47e9 Mon Sep 17 00:00:00 2001 From: Gautam Raj <156491596+gautam-rl@users.noreply.github.com> Date: Fri, 8 Aug 2025 13:46:17 -0700 Subject: [PATCH 2/2] Add jitter --- cmd/convertor/builder/builder_utils.go | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/cmd/convertor/builder/builder_utils.go b/cmd/convertor/builder/builder_utils.go index d6c83189..a6e9e2d7 100644 --- a/cmd/convertor/builder/builder_utils.go +++ b/cmd/convertor/builder/builder_utils.go @@ -25,6 +25,7 @@ import ( "fmt" "io" "math" + "math/rand" "os" "path" "time" @@ -88,8 +89,15 @@ func retryWithBackoff(ctx context.Context, maxRetries int, operation func() erro return lastErr } - // Exponential backoff: base delay of 1s, max 30s - backoffDelay := time.Duration(math.Min(float64(time.Second)*math.Pow(2, float64(attempt)), float64(30*time.Second))) + // Exponential backoff with random jitter: base delay of 1s, max 30s + baseDelay := time.Duration(math.Min(float64(time.Second)*math.Pow(2, float64(attempt)), float64(30*time.Second))) + // Add random jitter: ±25% of the base delay + jitter := time.Duration(float64(baseDelay) * (rand.Float64() - 0.5) * 0.5) + backoffDelay := baseDelay + jitter + // Ensure minimum delay of 500ms + if backoffDelay < 500*time.Millisecond { + backoffDelay = 500 * time.Millisecond + } logrus.Infof("received retryable error, retrying in %v (attempt %d/%d): %v", backoffDelay, attempt+1, maxRetries, lastErr) select {