OCPNODE-4477: docker: add blob-level mirror fallback in dockerImageSource

[QiWang19]

Summary

In disconnected/restricted environments, a mirror can pass the manifest
check in newImageSource() but then fail blob downloads (5xx, timeout,
or partial mirror returning BLOB_UNKNOWN). When this happens, the copy
fails and CRI-O propagates the error to kubelet — which retries from
scratch, picks the same broken mirror, and deadlocks.

This PR adds blob-level mirror fallback inside dockerImageSource so
that when a blob fetch fails, remaining configured mirrors are tried
before giving up. The fallback is transparent to all callers (CRI-O,
Podman, Buildah, Skopeo).

Design choices

Fallback triggers: Only transient errors (5xx, network timeout),
blob-not-found (BLOB_UNKNOWN), and rate limiting (429) trigger
fallback. Other 4xx errors (auth failures, etc.) are fatal — they
indicate a configuration problem, not a source issue.

Serialized mirror probing: getBlobWithMirrorFallback() holds
mirrorMu for the entire probe loop. This is intentional: the first
goroutine to hit a failure walks the mirror list while the rest block on
getActiveSource(), then reuse the discovered working mirror via
mirrorOverride. This avoids N goroutines independently exhausting the
mirror list.

No manifest re-fetch on fallback: Blob digests from the original
manifest are used against fallback sources. Safe because digests are
content-addressed. Fallback uses newPullClient() to create only a
client and reference, skipping the manifest fetch; configured mirrors
are expected to have the image in practice, and isMirrorFallbackError
handles BLOB_UNKNOWN to move to the next source if it doesn't.

remainingSources includes mirrors AND the primary location: The
fallback loop tries whatever pullSources entries come after the
initially-selected source, which may include the original upstream
registry.

Same-mirror retry on transient errors: tryGetBlob() retries once
with 1s+jitter delay before moving to the next source. This handles
brief 503 spikes without immediately triggering the full fallback
machinery.

Questions

Re: #845 (review)

• Are the right errors triggering fallback in isMirrorTransientError() / isMirrorFallbackError()? Should any other error types be included or excluded?
• The full-loop lock in getBlobWithMirrorFallback() serializes all blob goroutines during fallback — is this acceptable, or would you prefer a different concurrency strategy?
• Should tryGetBlob() retry once on the same mirror before falling back, or should any failure immediately try the next source?

Fixes: https://issues.redhat.com/browse/OCPSTRAT-3170

[mtrmac]

Thanks!

Doing this at the copy.Image level is very surprising, and harder than necessary — and less resilient, because it still assumes that a full image copy can proceed uninterrupted from a single mirror.

I’d expect dockerImageSource, instead, to do the fallbacks — making physicalRef and the dockerClient uses mutable (careful about HasThreadSafeGetBlob: true!), or perhaps maintaining an on-demand-extended set of dockerClients, in the worst case one per mirror; then every GetBlob/… on the dockerImageSource could fall back to another mirror (and we can have endless tuning discussions about the tuning logic there)…

Either way the generic copy code, AFAICS, does not need to know.

[mtrmac]

This is conceptually an internal state of dockerImageSource and I see no reason to make these public and store them so far from there. In the very worst case, we could use a new private interface for that (compare #531 adding NewImageSourceWithOptions) but even that seems suboptimal, see elsewhere.

[mtrmac]

The idea here is that we convert from types.ImageSource to private.ImageSource at the earliest opportunity, and then we never need to worry about public-only implementations (so much).

[mtrmac]

This seems unnecessary in principle … dockerImageSource already has a cachedManifest (but it does require maintain that state carefully).

[mtrmac]

BTW the non-timeout test failures look real at a first glance.

[QiWang19]

@mtrmac I updated this PR blob-level mirror fallback. I'd appreciate an early review on the approach before I continue. And there are a few design questions I'd like your input on(in the PR description under "Questions.")

[mtrmac]

Thanks!

I’m afraid there is way too much going on (e.g. we need to migrate of Cirrus) and I’ll be only able to pay proper attention to this, probably, weeks later.

For now just a very minimal look: Yes, I’d expect the state to be in dockerImageSource ~only. I didn’t carefully read the logic.

Are the right errors triggering fallback in isMirrorTransientError() / isMirrorFallbackError()?

Looks plausible; I’m sure we will be turning the heuristics 10 years from now… Anyway, thanks for clearly splitting the logic into separate documented functions, that will help future maintenance.

Should any other error types be included or excluded?

I think bodyReader already handles another set of errors. [There is a question whether bodyReader should also trigger a fallback to a different mirror… I think that would better be a different PR, and it might not even be necessary — OTOH there it might be worth just thinking a bit about what the code structure to enable that would look like.]

The full-loop lock in getBlobWithMirrorFallback() serializes all blob goroutines during fallback — is this acceptable, or would you prefer a different concurrency strategy?

Without looking at the details, I think this is a great idea.

Should tryGetBlob() retry once on the same mirror before falling back, or should any failure immediately try the next source?

Hum… in general, it’s problematic to have retry loops in lower levels of code; if each level of a call stack has a retry loop, that makes retry behavior exponential in the depth of the call stack. So as a general principle, I’ve been told that retries should exist only at the top level if possible. (Which is why GetBlob/GetManifest has no retries itself; callers can use c/common/pkg/retry at a higher level. OTOH bodyReader exists because the retry behavior in that case is usefully different from retrying from scratch.)

So I think GetBlob should ideally not retry at all if no mirrors are involved. If we do have mirrors, I don’t have a strong opinion. Given how costly / disruptive / potentially unexpected / (likely to fail in many configuration?) a fallback to another mirror can be, I think one more retry before starting the mirror fallback can well make sense.

[mtrmac]

Is there any other caller than GetBlob itself?

[QiWang19]

no other caller except 'GetBlob'. Make getBlob inlined.

[mtrmac]

Probably not?! If we got go the getBlob stage, cachedManifest is already set and ensure… does nothing.

[QiWang19]

Removed ensureManifestIsLoaded call in the fallnack loop.
ensureManifestIsLoaded was used to verify the fallback mirror had the image before trying blob fetches. Changed to just try the blob directly since most mirrors are expected to have the images in practice.

[QiWang19]

@mtrmac PR is ready for review. PTAL

[aguidirh]

Thanks for the PR @QiWang19.

I think there is a race condition between Close() and GetBlob()

The Problem

There's a race condition between Close() and concurrent GetBlob() calls that can cause blob downloads to fail with "use of closed network connection" errors.

What happens:

1. dockerImageSource declares HasThreadSafeGetBlob: true, allowing up to 6 concurrent blob downloads (see maxParallelDownloads in copy.go)
2. All 6 goroutines share the same HTTP client (s.c or s.mirrorOverride.client)
3. Each goroutine calls getActiveSource() to get the client pointer, then releases the lock before using it
4. If Close() is called while blobs are downloading, it closes the shared client while it's still in use

Timeline:

Thread A (GetBlob):                  Thread B (Close):
────────────────────────────────────────────────────────
1. Lock mirrorMu
2. Get client pointer
3. Unlock mirrorMu
4. Start blob download (slow!)
                                     5. Lock mirrorMu
                                     6. Copy client pointers
                                     7. Unlock mirrorMu
                                     8. Close all clients ← 💥
5. Blob download FAILS
   "use of closed network connection"

Impact

When downloading a typical container image with multiple layers:

• Up to 6 layers download in parallel, all using the same client
• One Close() call kills all active downloads simultaneously
• This is especially problematic for large images where layers take time to download

Code References

The issue is in these areas:

docker_image_src.go:247-262 - Close() doesn't wait for active blob fetches:

func (s *dockerImageSource) Close() error {
    s.mirrorMu.Lock()
    prev := s.prevOverrides
    override := s.mirrorOverride
    s.prevOverrides = nil
    s.mirrorOverride = nil
    s.mirrorMu.Unlock()  // Lock released here
    
    // But clients are closed AFTER releasing lock
    // while other goroutines may still be using them!
    for _, m := range prev {
        m.Close()
    }
    if override != nil {
        override.client.Close()
    }
    return s.c.Close()
}

docker_image_src.go:524-532 - getActiveSource() returns client pointer but doesn't prevent it from being closed:

func (s *dockerImageSource) getActiveSource() (*dockerClient, dockerReference, bool) {
    s.mirrorMu.Lock()
    defer s.mirrorMu.Unlock()
    if s.mirrorOverride != nil {
        return s.mirrorOverride.client, s.mirrorOverride.ref, hasRemaining
        // ^ Returns pointer, releases lock
        // Caller uses this pointer AFTER lock is released!
    }
    return s.c, s.physicalRef, hasRemaining
}

Suggested Fix

Add reference counting to track active blob fetches:

type dockerImageSource struct {
    // ... existing fields ...
    
    activeOps sync.WaitGroup  // Track active GetBlob operations
    closed    atomic.Bool     // Flag indicating Close() was called
}

func (s *dockerImageSource) GetBlob(...) {
    if s.closed.Load() {
        return nil, 0, errors.New("image source closed")
    }
    
    s.activeOps.Add(1)
    defer s.activeOps.Done()
    
    client, physRef, hasRemaining := s.getActiveSource()
    // ... rest of GetBlob ...
}

func (s *dockerImageSource) Close() error {
    s.closed.Store(true)
    s.activeOps.Wait()  // Wait for all active operations to complete
    
    // Now safe to close clients
    s.mirrorMu.Lock()
    prev := s.prevOverrides
    override := s.mirrorOverride
    s.prevOverrides = nil
    s.mirrorOverride = nil
    s.mirrorMu.Unlock()
    
    for _, m := range prev {
        m.Close()
    }
    if override != nil {
        override.client.Close()
    }
    return s.c.Close()
}

This ensures Close() waits for all in-flight blob downloads to complete before closing the clients.

[aguidirh]

Another code review round

[aguidirh]

Retry transient blob reads even when no mirrors remain.

Line 539 currently ties the one-shot retry to hasRemainingSources, so mirror-less pulls — and pulls already on the last source — fail on the first timeout/5xx instead of getting the transient retry this cohort adds.

Suggested change

	func tryGetBlob(ctx context.Context, client *dockerClient, physRef dockerReference,
	info types.BlobInfo, cache types.BlobInfoCache, hasRemainingSources bool,
	) (io.ReadCloser, int64, error) {
	reader, size, err := client.getBlob(ctx, physRef, info, cache)
	if err != nil && hasRemainingSources && isMirrorTransientError(err) {
	func tryGetBlob(ctx context.Context, client *dockerClient, physRef dockerReference,
	info types.BlobInfo, cache types.BlobInfoCache,
	) (io.ReadCloser, int64, error) {
	reader, size, err := client.getBlob(ctx, physRef, info, cache)
	if err != nil && isMirrorTransientError(err) {

[QiWang19]

So I think GetBlob should ideally not retry at all if no mirrors are involved. If we do have mirrors, I don’t have a strong opinion. Given how costly / disruptive / potentially unexpected / (likely to fail in many configuration?) a fallback to another mirror can be, I think one more retry before starting the mirror fallback can well make sense.

I think we can follow the previous comment from Miloslav, we can have a retry considering the disruptive mirror switch. if no mirror exist or we are on the last source, skipping the retry is intentional.

[aguidirh]

Keep probing later sources after a fallback candidate fails.

Lines 647-649 break out on the first non-classified blob error from a fallback source, and Line 668 then returns the stale originalErr. That can skip later sources that would have succeeded, including the primary, despite the new fallback contract/docs saying the remaining sources are tried in order.

Suggested change

	for len(s.remainingSources) > 0 {
	pullSource := s.remainingSources[0]
	s.remainingSources = s.remainingSources[1:]
	logrus.Debugf("Trying to access %q", pullSource.Reference)
	fallbackEndpoint, clientErr := newPullClient(s.fallbackSys, s.logicalRef, pullSource, registryConf)
	if clientErr != nil {
	logrus.Debugf("Accessing %q failed: %v", pullSource.Reference, clientErr)
	attempts = append(attempts, attempt{ref: pullSource.Reference, err: clientErr})
	continue
	}
	reader, size, err := tryGetBlob(ctx, fallbackEndpoint.client, fallbackEndpoint.ref, info, cache, len(s.remainingSources) > 0)
	if err != nil {
	fallbackEndpoint.client.Close()
	logrus.Debugf("Accessing %q failed: %v", pullSource.Reference, err)
	attempts = append(attempts, attempt{ref: pullSource.Reference, err: err})
	if !isMirrorTransientError(err) && !isMirrorFallbackError(err) {
	break
	}
	continue
	}
	if s.mirrorOverride != nil {
	s.prevOverrides = append(s.prevOverrides, s.mirrorOverride.client)
	}
	s.mirrorOverride = &mirrorSource{client: fallbackEndpoint.client, ref: fallbackEndpoint.ref}
	logrus.Debugf("Blob fetch succeeded from fallback source %q, switching to it for future requests", pullSource.Reference)
	return reader, size, nil
	}
	if len(attempts) > 0 {
	extras := []string{}
	for _, a := range attempts {
	extras = append(extras, fmt.Sprintf("[%s: %v]", a.ref.String(), a.err))
	}
	logrus.Debugf("(Fallback sources also failed: %s): %v", strings.Join(extras, "\n"), originalErr)
	}
	return nil, 0, originalErr
	finalErr := originalErr
	for len(s.remainingSources) > 0 {
	pullSource := s.remainingSources[0]
	s.remainingSources = s.remainingSources[1:]
	logrus.Debugf("Trying to access %q", pullSource.Reference)
	fallbackEndpoint, clientErr := newPullClient(s.fallbackSys, s.logicalRef, pullSource, registryConf)
	if clientErr != nil {
	logrus.Debugf("Accessing %q failed: %v", pullSource.Reference, clientErr)
	attempts = append(attempts, attempt{ref: pullSource.Reference, err: clientErr})
	continue
	}
	reader, size, err := tryGetBlob(ctx, fallbackEndpoint.client, fallbackEndpoint.ref, info, cache, len(s.remainingSources) > 0)
	if err != nil {
	fallbackEndpoint.client.Close()
	logrus.Debugf("Accessing %q failed: %v", pullSource.Reference, err)
	attempts = append(attempts, attempt{ref: pullSource.Reference, err: err})
	finalErr = err
	continue
	}
	if s.mirrorOverride != nil {
	s.prevOverrides = append(s.prevOverrides, s.mirrorOverride.client)
	}
	s.mirrorOverride = &mirrorSource{client: fallbackEndpoint.client, ref: fallbackEndpoint.ref}
	logrus.Debugf("Blob fetch succeeded from fallback source %q, switching to it for future requests", pullSource.Reference)
	return reader, size, nil
	}
	if len(attempts) > 0 {
	extras := []string{}
	for _, a := range attempts {
	extras = append(extras, fmt.Sprintf("[%s: %v]", a.ref.String(), a.err))
	}
	logrus.Debugf("(Fallback sources also failed: %s): %v", strings.Join(extras, "\n"), originalErr)
	}
	return nil, 0, finalErr

Also applies to: 661-668

[QiWang19]

I would defer to the maintainers on whether we should remove the break entirely. For the specific customer case that requests this feature, they care more about transient layer 7 errors from Quay:

    404  image not available
    502 bad gateway (overloaded or down for an upgrade)
    503 service unavailable
    QPS limits
    backend/storage issues

The goal of Lines 647-649 break out is to limit the scope of retry/fallback to 404, 429, 5xx, net.Error timeout error. Other errors may be cuaed by the image pull request it self, not mirror level, we break earlier to avoid the cost.

[mtrmac]

I think there is a race condition between Close() and GetBlob()

(An extremely brief drive-by, I didn’t read much further:) I think that should not generally happen: it’s clearly invalid to do in the single-threaded case, and I think that implies it is similarly invalid in the multi-threaded one. In practice, imageCopier.copyLayers would wait for all copies to finish before calling Close. Sure, being robust is valuable, but perhaps not to the point of overcomplicating the code for this.

[aguidirh]

I think there is a race condition between Close() and GetBlob()

(An extremely brief drive-by, I didn’t read much further:) I think that should not generally happen: it’s clearly invalid to do in the single-threaded case, and I think that implies it is similarly invalid in the multi-threaded one. In practice, imageCopier.copyLayers would wait for all copies to finish before calling Close. Sure, being robust is valuable, but perhaps not to the point of overcomplicating the code for this.

It the wait mechanism is already implemented there, so my suggestion about the race condition is not valid anymore. Please ignore it @QiWang19 and mark as solved.

[QiWang19]

rebased.