libindex: wire in httpreader.Reader

This handles the "easy" case of simply proxying reads for uncompressed
tar archives to range requests.

Future improvements would move the "spooling" out of this package and
into the `fs.FS` implementation.

Signed-off-by: Hank Donnay <hdonnay@redhat.com>
Change-Id: I9d200dd841954054df0b187b9fd160a56a6a6964

Author: hdonnay

internal/httpreader/reader.go

@@ -88,8 +88,14 @@ var (
 
 // ReadAt implements [io.ReaderAt].
 func (r *Reader) ReadAt(b []byte, off int64) (int, error) {
-	if len(b) == 0 {
+	switch {
+	case len(b) == 0:
+		// Nothing to read.
 		return 0, nil
+	case off >= r.size:
+		// Impossible to read anything, so copy [os.File] behavior.
+		// Use ≥ because it's comparing an index and a length.
+		return 0, io.EOF
 	}
 
 	req, err := http.NewRequestWithContext(r.ctx, http.MethodGet, r.res, nil)
@@ -105,6 +111,8 @@ func (r *Reader) ReadAt(b []byte, off int64) (int, error) {
 	switch res.StatusCode {
 	case http.StatusPartialContent: // OK
 	case http.StatusOK:
+		// NOTE(hank) I've seen servers in the wild that just send the file if
+		// the range starts at 0, even if they support range requests.
 		if off != 0 {
 			return 0, errWithStatus(srvBotch(req.Header.Get(`range`)), res.StatusCode)
 		}

libindex/fetcher.go

@@ -9,20 +9,24 @@ import (
 	"fmt"
 	"io"
 	"log/slog"
+	"mime"
 	"net/http"
 	"net/url"
 	"os"
+	"regexp"
 	"runtime"
 	"strings"
 
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/codes"
+	semconv "go.opentelemetry.io/otel/semconv/v1.41.0"
 	"go.opentelemetry.io/otel/trace"
 	"golang.org/x/sync/errgroup"
 
 	"github.com/quay/claircore"
 	"github.com/quay/claircore/indexer"
 	"github.com/quay/claircore/internal/cache"
+	"github.com/quay/claircore/internal/httpreader"
 	"github.com/quay/claircore/internal/httputil"
 	"github.com/quay/claircore/internal/wart"
 	"github.com/quay/claircore/internal/zreader"
@@ -125,6 +129,43 @@ func (a *RemoteFetchArena) fetchInto(ctx context.Context, l *claircore.Layer, cl
 			span.End()
 		}()
 
+		var d details
+		d, err = a.inspect(ctx, desc)
+		if err != nil {
+			return err
+		}
+
+	HTTPReader:
+		switch { // Switch for a single case to be able to break.
+		case d.Uncompressed:
+			var opts []httpreader.Option
+			if d.ContentLength > 0 {
+				opts = append(opts, httpreader.WithSize(d.ContentLength))
+			}
+			if len(desc.Headers) != 0 {
+				opts = append(opts, httpreader.WithHeaders(desc.Headers))
+			}
+			var rd *httpreader.Reader
+			rd, err = httpreader.New(ctx, a.wc, desc.URI, opts...)
+			switch {
+			case err == nil:
+				if err = l.Init(ctx, desc, rd); err != nil {
+					return err
+				}
+				*cl = closeFunc(func() (err error) {
+					err = errors.Join(l.Close(), rd.Close())
+					return err
+				})
+				a.logger(desc).DebugContext(ctx, "using httpreader")
+				return nil
+			case errors.Is(err, errors.ErrUnsupported):
+				err = nil
+				break HTTPReader
+			default:
+				return err
+			}
+		}
+
 		// NB This is not closed on purpose. The [io.Closer] populated by this
 		// function holds the pointer until that function is cleaned up. Once
 		// nothing has a copy of this [*os.File], the runtime will run all the
@@ -135,19 +176,20 @@ func (a *RemoteFetchArena) fetchInto(ctx context.Context, l *claircore.Layer, cl
 		var spool *os.File
 		spool, err = a.files.Get(ctx, key, func(ctx context.Context, _ string) (*os.File, error) {
 			cacheHit = false
-			return a.fetchFileForCache(ctx, desc)
+			return a.fetchFileForCache(ctx, desc, d)
 		})
 		if err != nil {
 			return err
 		}
+		var f *os.File
 		// This is an owned, independent descriptor for the passed [*os.File].
-		f, err := reopen(a.root, spool)
+		f, err = reopen(a.root, spool)
 		if err != nil {
 			return err
 		}
 
 		// If this succeeds, "f" is now owned by "l"
-		if err := l.Init(ctx, desc, f); err != nil {
+		if err = l.Init(ctx, desc, f); err != nil {
 			return errors.Join(err, f.Close())
 		}
 		*cl = closeFunc(func() (err error) {
@@ -173,12 +215,106 @@ func (f closeFunc) Close() error {
 	return f()
 }
 
+// Inspect makes a request to the layer's URI and examines the response for
+// useful information.
+func (a *RemoteFetchArena) inspect(ctx context.Context, desc *claircore.LayerDescription) (d details, err error) {
+	ctx, span := tracer.Start(ctx, "RemoteFetchArena.inspect")
+	defer func() {
+		a.logger(desc).DebugContext(ctx, "inspected resource", "ok", err == nil, "details", &d)
+		span.RecordError(err)
+		span.End()
+	}()
+	span.SetStatus(codes.Error, "")
+
+	var req *http.Request
+	var res *http.Response
+	req, err = http.NewRequestWithContext(ctx, http.MethodGet, desc.URI, nil)
+	if err != nil {
+		return d, fmt.Errorf("fetcher: failed to construct request: %w", err)
+	}
+	req.Header = http.Header(desc.Headers).Clone()
+	req.Header.Set(`claircore-reason`, `inspect`)
+	req.Header.Set(`range`, `bytes=0-15`)
+	res, err = a.wc.Do(req)
+	if err != nil {
+		return d, fmt.Errorf("fetcher: request failed: %w", err)
+	}
+	err = httputil.CheckResponse(res, http.StatusOK, http.StatusPartialContent)
+	if err != nil {
+		return d, fmt.Errorf("fetcher: %w", err)
+	}
+	head := make([]byte, 16)
+	_, err = io.ReadFull(res.Body, head)
+	_ = res.Body.Close()
+	if err != nil {
+		return d, fmt.Errorf("fetcher: unexpected read: %w", err)
+	}
+
+	const (
+		ctKey = `content-type`
+		arKey = `accept-ranges`
+	)
+	ct, _, err := mime.ParseMediaType(res.Header.Get(ctKey))
+	if err != nil {
+		return d, fmt.Errorf("fetcher: %w", err)
+	}
+	span.SetAttributes(
+		semconv.HTTPResponseStatusCode(res.StatusCode),
+		semconv.HTTPResponseBodySize(int(res.ContentLength)),
+		semconv.HTTPResponseHeader(ctKey, ct),
+		semconv.HTTPResponseHeader(arKey, res.Header.Get(arKey)),
+	)
+	if isOctetStream(ct) && zreader.DetectCompression(head) == zreader.KindNone {
+		ct = `application/x-tar`
+	}
+	switch res.StatusCode {
+	case http.StatusOK:
+		d.ContentLength = res.ContentLength
+	case http.StatusPartialContent:
+		var cr httpreader.ContentRange
+		if err := cr.Parse(res.Header.Get(`content-range`)); err == nil {
+			d.ContentLength = cr.Length
+		}
+	}
+	d.Uncompressed = ct == "application/x-tar" || strings.HasSuffix(ct, ".tar")
+	d.RangeOK = res.Header.Get(arKey) == `bytes` || res.StatusCode == http.StatusPartialContent
+
+	span.SetStatus(codes.Ok, "")
+	return d, nil
+}
+
+var _ slog.LogValuer = (*details)(nil)
+
+// Details is the useful information reported by [RemoteFetchArena.inspect].
+type details struct {
+	// The content length, if known.
+	ContentLength int64
+	// The resource reports to be known non-compressed contents.
+	Uncompressed bool
+	// The server reports supporting "bytes" via the "Accept-Ranges" header.
+	RangeOK bool
+}
+
+// LogValue implements [slog.LogValuer].
+func (d *details) LogValue() slog.Value {
+	return slog.GroupValue(
+		slog.Int64("content-length", d.ContentLength),
+		slog.Bool("uncompressed", d.Uncompressed),
+		slog.Bool("range-ok", d.RangeOK),
+	)
+}
+
+// Logger returns a [*slog.Logger] with predefined attributes.
+func (a *RemoteFetchArena) logger(desc *claircore.LayerDescription) *slog.Logger {
+	return slog.With("arena", a.root.Name(), "layer", desc.Digest, "uri", desc.URI)
+}
+
 // FetchFileForCache is the inner function used inside the [cache.Live].
 //
 // Because we know we're the only concurrent call that's dealing with this blob,
 // we can be a bit more lax.
-func (a *RemoteFetchArena) fetchFileForCache(ctx context.Context, desc *claircore.LayerDescription) (*os.File, error) {
-	log := slog.With("arena", a.root.Name(), "layer", desc.Digest, "uri", desc.URI)
+func (a *RemoteFetchArena) fetchFileForCache(ctx context.Context, desc *claircore.LayerDescription, _ details) (*os.File, error) {
+	log := a.logger(desc)
 	ctx, span := tracer.Start(ctx, "RemoteFetchArena.fetchFileForCache")
 	defer span.End()
 	span.SetStatus(codes.Error, "")
@@ -210,6 +346,7 @@ func (a *RemoteFetchArena) fetchFileForCache(ctx context.Context, desc *claircor
 		URL:        url,
 		Header:     http.Header(desc.Headers).Clone(),
 	}).WithContext(ctx)
+	req.Header.Set(`claircore-reason`, `fetch`)
 	resp, err := a.wc.Do(req)
 	if err != nil {
 		return nil, fmt.Errorf("fetcher: request failed: %w", err)
@@ -235,10 +372,13 @@ func (a *RemoteFetchArena) fetchFileForCache(ctx context.Context, desc *claircor
 	}
 	defer zr.Close()
 	// Look at the content-type and optionally fix it up.
-	ct := resp.Header.Get("content-type")
+	ct, _, err := mime.ParseMediaType(resp.Header.Get("content-type"))
+	if err != nil {
+		return nil, fmt.Errorf("fetcher: %w", err)
+	}
 	log.DebugContext(ctx, "reported content-type", "content-type", ct)
 	span.SetAttributes(payloadType(ct), payloadCompression(kind))
-	if ct == "" || ct == "text/plain" || ct == "binary/octet-stream" || ct == "application/octet-stream" {
+	if ct == "" || ct == "text/plain" || isOctetStream(ct) {
 		switch kind {
 		case zreader.KindGzip:
 			ct = "application/gzip"
@@ -420,3 +560,14 @@ func (p *FetchProxy) Close() error {
 	}
 	return errors.Join(errs...)
 }
+
+// IsOctetStream tests if the incoming media type string is an "octet-stream"
+// type.
+//
+// Reports false for strings that are not of the form "<type>/<subtype>". For
+// interoperability purposes, the nonstandard "binary" type is accepted.
+func isOctetStream(ct string) bool {
+	return reOctetStream.MatchString(ct)
+}
+
+var reOctetStream = regexp.MustCompile(`^(application|binary)/octet-stream$`)