diff --git a/designs/gantry-128gib-single-layer-pulls.md b/designs/gantry-128gib-single-layer-pulls.md index 03c54611d..9a1cf96d5 100644 --- a/designs/gantry-128gib-single-layer-pulls.md +++ b/designs/gantry-128gib-single-layer-pulls.md @@ -110,9 +110,9 @@ resumable. | ID | Issue | Evidence | Owning PR | Status | |---|---|---|---|---| -| L4 | Gantry origin pulls do not send `Range` and cannot request an origin body from an offset. | `internal/gantry/origin/origin.go:25`, `internal/gantry/origin/origin.go:477-542` | PR 4 | Open | +| L4 | Gantry origin pulls do not send `Range` and cannot request an origin body from an offset. | `internal/gantry/origin/origin.go:25`, `internal/gantry/origin/origin.go:477-542` | PR 4 | Addressed | | L5 | A failed background ingest is aborted. A later writer with a stale nonzero offset is also aborted because callers restart at byte zero. | `internal/gantry/containerdstore/store.go:328-383`, `internal/gantry/containerdstore/store.go:502-519` | PR 5 | Open | -| L6 | Gantry does not handle the local containerd client's inbound `Range` header on the origin path. | `internal/gantry/mirror/mirror.go:685-930` | PR 4 | Open | +| L6 | Gantry does not handle the local containerd client's inbound `Range` header on the origin path. | `internal/gantry/mirror/mirror.go:685-930` | PR 4 | Addressed | L4-L6 do not cause the fixed five- or 30-minute failures. They determine whether a transfer interrupted for another reason can continue without @@ -289,13 +289,13 @@ PR reports. **Purpose:** Avoid replaying the origin prefix when local containerd retries a live Gantry response at a nonzero offset. -- [ ] Validate a single inbound blob range from local containerd. -- [ ] Carry the offset through `ifaces.OriginRef` into the origin request. -- [ ] Require and validate the origin's `206` and `Content-Range` response. -- [ ] Return matching range semantics to containerd. -- [ ] Preserve request-scoped delegated registry authorization on the origin +- [x] Validate a single inbound blob range from local containerd. +- [x] Carry the offset through `ifaces.OriginRef` into the origin request. +- [x] Require and validate the origin's `206` and `Content-Range` response. +- [x] Return matching range semantics to containerd. +- [x] Preserve request-scoped delegated registry authorization on the origin retry and never forward it to a peer. -- [ ] Test supported ranges, ignored ranges, malformed ranges, and a mid-body +- [x] Test supported ranges, ignored ranges, malformed ranges, and an offset interruption followed by an offset retry. **Addresses:** L4 and L6. diff --git a/internal/gantry/ifaces/ifaces.go b/internal/gantry/ifaces/ifaces.go index 7f26a370f..bcd6c2c5c 100644 --- a/internal/gantry/ifaces/ifaces.go +++ b/internal/gantry/ifaces/ifaces.go @@ -128,7 +128,7 @@ type OriginRef struct { Repository string // e.g. "library/nginx" Digest digest.Digest // Offset requests bytes starting at this position when fetching from a - // peer. Origin registry callers ignore it. Zero requests the full object. + // peer or origin registry. Zero requests the full object. Offset int64 // Kind discriminates the OCI Distribution Spec URL family for this diff --git a/internal/gantry/mirror/mirror.go b/internal/gantry/mirror/mirror.go index 2ffd9cb6b..6bd21a52c 100644 --- a/internal/gantry/mirror/mirror.go +++ b/internal/gantry/mirror/mirror.go @@ -850,6 +850,15 @@ func (s *Server) serveDigest(w http.ResponseWriter, r *http.Request, upstream, r slog.String("kind", kind.String()), ) + if rangeHeader := r.Header.Get("Range"); rangeHeader != "" { + offset, ok := parseOriginRetryRange(rangeHeader) + if ok && kind == ifaces.KindBlob { + s.serveOriginRange(ctx, w, d, kind, upstream, repo, offset, logger) + + return + } + } + // 1. Local content-store lookup. if handled := s.serveLocalHit(ctx, w, r, d, kind, upstream, repo, logger); handled { return @@ -925,6 +934,103 @@ func (s *Server) serveDigest(w http.ResponseWriter, r *http.Request, upstream, r s.serveFromOrigin(ctx, w, d, kind, upstream, repo, logger) } +func parseOriginRetryRange(value string) (int64, bool) { + const prefix = "bytes=" + if !strings.HasPrefix(value, prefix) || strings.Contains(value, ",") { + return 0, false + } + + spec := strings.TrimPrefix(value, prefix) + if !strings.HasSuffix(spec, "-") { + return 0, false + } + + offset, err := strconv.ParseInt(strings.TrimSuffix(spec, "-"), 10, 64) + if err != nil || offset <= 0 { + return 0, false + } + + return offset, true +} + +func (s *Server) serveOriginRange(ctx context.Context, w http.ResponseWriter, d digest.Digest, kind ifaces.OriginRefKind, upstream, repo string, offset int64, logger *slog.Logger) { + pullStartedAt := time.Now() + ref := ifaces.OriginRef{Registry: upstream, Repository: repo, Digest: d, Offset: offset, Kind: kind} + + if s.liveStreamThrough { + s.fireOriginStreamStarted(kind) + } + + rc, totalSize, err := s.origin.Pull(ctx, ref) + if err != nil { + logger.Debug("mirror: origin range pull failed", + slog.String("pull_mode", "live_range"), + slog.String("deadline_owner", originDeadlineOwner(ctx, err)), + slog.Duration("elapsed", time.Since(pullStartedAt)), + slog.Int64("expected_size", -1), + slog.Int64("written", 0), + slog.Int64("offset", offset), + slog.Any("err", err), + ) + + if s.liveStreamThrough { + s.fireOriginStreamFailed(kind) + } + + writeOriginError(w, err, logger) + + return + } + + defer func() { _ = rc.Close() }() //nolint:errcheck // best-effort close + + remaining := totalSize - offset + if remaining <= 0 { + if s.liveStreamThrough { + s.fireOriginStreamFailed(kind) + } + + http.Error(w, "invalid origin range size", http.StatusBadGateway) + + return + } + + w.Header().Set("Accept-Ranges", "bytes") + w.Header().Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", offset, totalSize-1, totalSize)) + w.Header().Set("Content-Length", strconv.FormatInt(remaining, 10)) + writeBlobHeaders(w, d, -1, kind) + w.WriteHeader(http.StatusPartialContent) + + written, copyErr := streamcopy.CopyN(w, rc, remaining) + s.fireMirrorBytesServed(kind, "origin", written) + + if copyErr != nil { + logger.Debug("mirror: live origin range stream failed", + slog.String("pull_mode", "live_range"), + slog.String("deadline_owner", originDeadlineOwner(ctx, copyErr)), + slog.Duration("elapsed", time.Since(pullStartedAt)), + slog.Int64("expected_size", totalSize), + slog.Int64("written", written), + slog.Int64("offset", offset), + slog.Any("err", copyErr), + ) + + if s.liveStreamThrough { + s.fireOriginStreamFailed(kind) + } + + return + } + + if s.liveStreamThrough { + s.fireOriginStreamCompleted(kind) + } + + s.fireMirrorResponseCompleted(d, kind, "origin") + s.fireLiveStreamCompleted(d) + s.recordNegCacheSuccess(d) +} + // serveLocalHit serves d from the local content store when present. // Returns true when the response has been fully written (cache hit or // non-NotFound store error -> 5xx); false means the digest is diff --git a/internal/gantry/mirror/mirror_test.go b/internal/gantry/mirror/mirror_test.go index 9f7b4cafd..0229e1c9b 100644 --- a/internal/gantry/mirror/mirror_test.go +++ b/internal/gantry/mirror/mirror_test.go @@ -146,6 +146,30 @@ type authorizationCapturingOrigin struct { seen chan string } +type rangeOriginRequest struct { + offset int64 + authorization string +} + +type rangeCapturingOrigin struct { + body []byte + seen chan rangeOriginRequest +} + +func (o *rangeCapturingOrigin) Pull(ctx context.Context, ref ifaces.OriginRef) (io.ReadCloser, int64, error) { + o.seen <- rangeOriginRequest{offset: ref.Offset, authorization: registryauth.Authorization(ctx)} + + if ref.Offset < 0 || ref.Offset >= int64(len(o.body)) { + return nil, 0, errors.New("invalid offset") + } + + return io.NopCloser(bytes.NewReader(o.body[ref.Offset:])), int64(len(o.body)), nil +} + +func (o *rangeCapturingOrigin) Head(context.Context, ifaces.OriginRef) (int64, string, error) { + return int64(len(o.body)), "application/octet-stream", nil +} + type authorizationRejectingOrigin struct{} func (authorizationRejectingOrigin) Pull(_ context.Context, ref ifaces.OriginRef) (io.ReadCloser, int64, error) { @@ -168,6 +192,101 @@ func (o *authorizationCapturingOrigin) Head(ctx context.Context, _ ifaces.Origin return int64(len(o.body)), "application/octet-stream", nil } +func TestMirrorOriginRangeRetry(t *testing.T) { + body := []byte("0123456789") + d := digestOf(body) + origin := &rangeCapturingOrigin{body: body, seen: make(chan rangeOriginRequest, 1)} + cfg := &config.Config{UpstreamRegistries: []config.UpstreamRegistry{{Name: "reg.example.com", Endpoint: "https://reg.example.com"}}} + + srv := httptest.NewServer(mirror.New(cfg, fakes.NewCache(), origin, mirror.WithLiveStreamThrough()).Handler()) + defer srv.Close() + + req, err := http.NewRequest(http.MethodGet, srv.URL+"/v2/repo/blobs/"+d.String(), nil) + if err != nil { + t.Fatal(err) + } + + req.Header.Set("Range", "bytes=4-") + req.Header.Set("Authorization", "Bearer requester-token") + + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusPartialContent { + t.Fatalf("status = %d; want 206", resp.StatusCode) + } + + if got := resp.Header.Get("Content-Range"); got != "bytes 4-9/10" { + t.Fatalf("Content-Range = %q; want bytes 4-9/10", got) + } + + got, err := io.ReadAll(resp.Body) + if err != nil { + t.Fatal(err) + } + + if string(got) != string(body[4:]) { + t.Fatalf("body = %q; want %q", got, body[4:]) + } + + seen := <-origin.seen + if seen.offset != 4 { + t.Fatalf("origin offset = %d; want 4", seen.offset) + } + + if seen.authorization != "Bearer requester-token" { + t.Fatalf("origin authorization = %q; want requester token", seen.authorization) + } +} + +func TestMirrorPreservesFullBodyPathForUnsupportedOriginRange(t *testing.T) { + for _, rangeHeader := range []string{"bytes=0-", "bytes=4-6", "invalid"} { + t.Run(rangeHeader, func(t *testing.T) { + body := []byte("0123456789") + d := digestOf(body) + origin := &rangeCapturingOrigin{body: body, seen: make(chan rangeOriginRequest, 1)} + cfg := &config.Config{UpstreamRegistries: []config.UpstreamRegistry{{Name: "reg.example.com", Endpoint: "https://reg.example.com"}}} + + srv := httptest.NewServer(mirror.New(cfg, fakes.NewCache(), origin).Handler()) + defer srv.Close() + + req, err := http.NewRequest(http.MethodGet, srv.URL+"/v2/repo/blobs/"+d.String(), nil) + if err != nil { + t.Fatal(err) + } + + req.Header.Set("Range", rangeHeader) + + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + t.Fatalf("status = %d; want 200", resp.StatusCode) + } + + got, err := io.ReadAll(resp.Body) + if err != nil { + t.Fatal(err) + } + + if string(got) != string(body) { + t.Fatalf("body = %q; want %q", got, body) + } + + seen := <-origin.seen + if seen.offset != 0 { + t.Fatalf("origin offset = %d; want 0", seen.offset) + } + }) + } +} + func TestMirror_CapturesInboundAuthorizationForOrigin(t *testing.T) { body := []byte("origin bytes") d := digestOf(body) diff --git a/internal/gantry/origin/origin.go b/internal/gantry/origin/origin.go index 57d02d8a0..cd3b5871a 100644 --- a/internal/gantry/origin/origin.go +++ b/internal/gantry/origin/origin.go @@ -22,7 +22,6 @@ // // - the design doc negative-cache cooldown integration . // - Per-pull retries with backoff (caller's responsibility for now). -// - Resumable / ranged pulls (the design doc layer-pull semantics). package origin import ( @@ -482,14 +481,18 @@ func (r *registry) rememberAuthenticationChallenge(challenge string) { } func (r *registry) pull(ctx context.Context, ref ifaces.OriginRef) (io.ReadCloser, int64, error) { + if ref.Offset < 0 { + return nil, 0, &ifaces.OriginError{Ref: ref, Class: ifaces.FailureTransient, Err: fmt.Errorf("origin offset %d is negative", ref.Offset)} + } + path := r.urlFor(ref) - resp, err := r.do(ctx, http.MethodGet, path) + resp, err := r.do(ctx, http.MethodGet, path, ref.Offset) if err != nil { return nil, 0, &ifaces.OriginError{Ref: ref, Class: classOf(err), Err: err} } - if resp.StatusCode == http.StatusNotFound && ref.Kind != ifaces.KindManifest { + if resp.StatusCode == http.StatusNotFound && ref.Kind != ifaces.KindManifest && ref.Offset == 0 { // Containerd treats every digest in a pod spec as a generic // "content descriptor" and fetches it via /v2//blobs/ // . When that digest happens to be an image manifest, @@ -509,7 +512,7 @@ func (r *registry) pull(ctx context.Context, ref ifaces.OriginRef) (io.ReadClose mRef.Kind = ifaces.KindManifest mPath := r.urlFor(mRef) - mResp, mErr := r.do(ctx, http.MethodGet, mPath) + mResp, mErr := r.do(ctx, http.MethodGet, mPath, 0) if mErr == nil && mResp.StatusCode == http.StatusOK { mSize := int64(-1) @@ -546,6 +549,24 @@ func (r *registry) pull(ctx context.Context, ref ifaces.OriginRef) (io.ReadClose return nil, 0, &ifaces.OriginError{Ref: ref, Class: ifaces.FailureNotFound, Err: fmt.Errorf("origin: %s not found (blob 404, manifest 404)", ref.Digest)} } + if ref.Offset > 0 { + if resp.StatusCode != http.StatusPartialContent { + defer func() { _ = resp.Body.Close() }() //nolint:errcheck // best-effort body close + + return nil, 0, &ifaces.OriginError{Ref: ref, Class: ifaces.FailureTransient, Err: fmt.Errorf("origin ignored range offset %d: status %s", ref.Offset, resp.Status)} + } + + start, end, size, ok := parseOriginContentRange(resp.Header.Get("Content-Range")) + if !ok || start != ref.Offset || end != size-1 || + (resp.ContentLength >= 0 && resp.ContentLength != end-start+1) { + defer func() { _ = resp.Body.Close() }() //nolint:errcheck // best-effort body close + + return nil, 0, &ifaces.OriginError{Ref: ref, Class: ifaces.FailureTransient, Err: fmt.Errorf("origin returned invalid Content-Range %q for offset %d", resp.Header.Get("Content-Range"), ref.Offset)} + } + + return resp.Body, size, nil + } + if resp.StatusCode != http.StatusOK { defer func() { _ = resp.Body.Close() }() //nolint:errcheck // best-effort body close return nil, 0, r.classify(ref, resp) @@ -580,7 +601,7 @@ func (r *registry) pull(ctx context.Context, ref ifaces.OriginRef) (io.ReadClose func (r *registry) head(ctx context.Context, ref ifaces.OriginRef) (int64, string, error) { path := r.urlFor(ref) - resp, err := r.do(ctx, http.MethodHead, path) + resp, err := r.do(ctx, http.MethodHead, path, 0) if err != nil { return 0, "", &ifaces.OriginError{Ref: ref, Class: classOf(err), Err: err} } @@ -590,7 +611,7 @@ func (r *registry) head(ctx context.Context, ref ifaces.OriginRef) (int64, strin mRef := ref mRef.Kind = ifaces.KindManifest - mResp, mErr := r.do(ctx, http.MethodHead, r.urlFor(mRef)) + mResp, mErr := r.do(ctx, http.MethodHead, r.urlFor(mRef), 0) if mErr == nil && mResp.StatusCode == http.StatusOK { defer func() { _ = mResp.Body.Close() }() //nolint:errcheck // best-effort body close @@ -654,7 +675,7 @@ func (r *registry) urlFor(ref ifaces.OriginRef) string { // is rejected, the 401 is returned rather than silently changing identity to // this node's configured credentials. Without delegated auth, the legacy // credentials-file bearer-token flow remains available. -func (r *registry) do(ctx context.Context, method, urlStr string) (*http.Response, error) { +func (r *registry) do(ctx context.Context, method, urlStr string, offset int64) (*http.Response, error) { delegatedAuthorization := registryauth.Authorization(ctx) if delegatedAuthorization != "" && !r.canSendBasicAuth() { return nil, &tokenError{ @@ -681,6 +702,10 @@ func (r *registry) do(ctx context.Context, method, urlStr string) (*http.Respons req.Header.Set("Authorization", authorization) } + if offset > 0 { + req.Header.Set("Range", fmt.Sprintf("bytes=%d-", offset)) + } + return req, nil } @@ -723,7 +748,7 @@ func (r *registry) do(ctx context.Context, method, urlStr string) (*http.Respons if !strings.HasPrefix(strings.ToLower(challenge), "bearer ") { // No bearer challenge - return 401 verbatim so classify reports auth. - return r.repeatWithoutToken(ctx, method, urlStr) + return r.repeatWithoutToken(ctx, method, urlStr, offset) } tok, ttl, err := r.fetchBearerToken(ctx, challenge) @@ -744,7 +769,7 @@ func (r *registry) do(ctx context.Context, method, urlStr string) (*http.Respons // repeatWithoutToken re-issues a request that received a 401 but no usable // bearer challenge. Returns the 401 response so the caller can classify it // as FailureAuth. -func (r *registry) repeatWithoutToken(ctx context.Context, method, urlStr string) (*http.Response, error) { +func (r *registry) repeatWithoutToken(ctx context.Context, method, urlStr string, offset int64) (*http.Response, error) { req, err := http.NewRequestWithContext(ctx, method, urlStr, nil) if err != nil { return nil, err @@ -754,9 +779,46 @@ func (r *registry) repeatWithoutToken(ctx context.Context, method, urlStr string req.SetBasicAuth(r.username, r.password) } + if offset > 0 { + req.Header.Set("Range", fmt.Sprintf("bytes=%d-", offset)) + } + return r.hc.Do(req) } +func parseOriginContentRange(value string) (start, end, size int64, ok bool) { + if !strings.HasPrefix(value, "bytes ") { + return 0, 0, 0, false + } + + rangeAndSize := strings.Split(strings.TrimPrefix(value, "bytes "), "/") + if len(rangeAndSize) != 2 { + return 0, 0, 0, false + } + + bounds := strings.Split(rangeAndSize[0], "-") + if len(bounds) != 2 { + return 0, 0, 0, false + } + + start, err := strconv.ParseInt(bounds[0], 10, 64) + if err != nil { + return 0, 0, 0, false + } + + end, err = strconv.ParseInt(bounds[1], 10, 64) + if err != nil { + return 0, 0, 0, false + } + + size, err = strconv.ParseInt(rangeAndSize[1], 10, 64) + if err != nil { + return 0, 0, 0, false + } + + return start, end, size, start >= 0 && end >= start && size > end +} + // fetchBearerToken parses a Bearer challenge and exchanges it for a token. // Returns the token and the server-advertised TTL (or 0 if the response // omitted expires_in, in which case the caller picks a default). diff --git a/internal/gantry/origin/origin_test.go b/internal/gantry/origin/origin_test.go index 858290552..57cbad550 100644 --- a/internal/gantry/origin/origin_test.go +++ b/internal/gantry/origin/origin_test.go @@ -125,6 +125,83 @@ func TestPullBlob_Success(t *testing.T) { } } +func TestPullBlobRange(t *testing.T) { + body := []byte("0123456789") + d := digestOf(body) + + const offset = int64(4) + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if got := r.Header.Get("Range"); got != "bytes=4-" { + t.Errorf("Range = %q; want bytes=4-", got) + } + + w.Header().Set("Content-Length", "6") + w.Header().Set("Content-Range", "bytes 4-9/10") + w.WriteHeader(http.StatusPartialContent) + _, _ = w.Write(body[offset:]) //nolint:errcheck // best-effort write + })) + defer srv.Close() + + c := newClient(t, config.UpstreamRegistry{Name: "reg", Endpoint: srv.URL}) + + rc, size, err := c.Pull(context.Background(), ifaces.OriginRef{ + Registry: "reg", Repository: "library/nginx", Digest: d, Kind: ifaces.KindBlob, Offset: offset, + }) + if err != nil { + t.Fatalf("Pull: %v", err) + } + defer rc.Close() + + got, err := io.ReadAll(rc) + if err != nil { + t.Fatalf("ReadAll: %v", err) + } + + if string(got) != string(body[offset:]) { + t.Fatalf("body = %q; want %q", got, body[offset:]) + } + + if size != int64(len(body)) { + t.Fatalf("size = %d; want %d", size, len(body)) + } +} + +func TestPullBlobRangeRejectsIgnoredOrInvalidResponse(t *testing.T) { + body := []byte("0123456789") + d := digestOf(body) + + tests := []struct { + name string + status int + contentRange string + want string + }{ + {name: "ignored", status: http.StatusOK, want: "ignored range"}, + {name: "invalid", status: http.StatusPartialContent, contentRange: "bytes 0-5/10", want: "invalid Content-Range"}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Range", test.contentRange) + w.WriteHeader(test.status) + _, _ = w.Write(body) //nolint:errcheck // best-effort write + })) + defer srv.Close() + + c := newClient(t, config.UpstreamRegistry{Name: "reg", Endpoint: srv.URL}) + + _, _, err := c.Pull(context.Background(), ifaces.OriginRef{ + Registry: "reg", Repository: "library/nginx", Digest: d, Kind: ifaces.KindBlob, Offset: 4, + }) + if err == nil || !strings.Contains(err.Error(), test.want) { + t.Fatalf("error = %v; want %q", err, test.want) + } + }) + } +} + func TestPullManifest_AcceptHeaderAndPath(t *testing.T) { body := []byte(`{"schemaVersion":2}`) d := digestOf(body)