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

libindex/fetcher.go

@@ -3,26 +3,31 @@ package libindex
 import (
 	"bufio"
 	"bytes"
+	"cmp"
 	"context"
 	"encoding/hex"
 	"errors"
 	"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"
@@ -131,40 +136,85 @@ func (a *RemoteFetchArena) fetchInto(ctx context.Context, l *claircore.Layer, cl
 		// cleanup logic associated with the pointer.
 		//
 		// Every new [*claircore.Layer] gets its own file descriptor via the
-		// [reopen] helper.
+		// [reopen] helper, or eschews using a file at all.
 		var spool *os.File
+		var forceFetch bool
+		var d details
+	Fetch:
 		spool, err = a.files.Get(ctx, key, func(ctx context.Context, _ string) (*os.File, error) {
 			cacheHit = false
-			return a.fetchFileForCache(ctx, desc)
+			d, err = a.inspect(ctx, desc)
+			if err != nil {
+				return nil, err
+			}
+			if !forceFetch && d.Uncompressed && d.RangeOK {
+				return nil, errUseHTTPReader
+			}
+			return a.fetchFileForCache(ctx, desc, d)
 		})
-		if err != nil {
-			return err
-		}
-		// This is an owned, independent descriptor for the passed [*os.File].
-		f, err := reopen(a.root, spool)
-		if err != nil {
-			return err
-		}
+		switch err {
+		case nil: // Reopen the returned [*os.File].
+			var f *os.File
+			// This is an owned, independent descriptor for the passed [*os.File].
+			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 {
-			return errors.Join(err, f.Close())
-		}
-		*cl = closeFunc(func() (err error) {
-			err = errors.Join(l.Close(), f.Close())
-			// Using this KeepAlive keeps the cached file descriptor live until
-			// all users of the blob have cleaned up. This should be after "f"
-			// is closed so that the cached-owned file descriptor outlives any
-			// reopened copies. There's no explicit association of these file
-			// descriptors, it's all kernel-side book-keeping.
-			runtime.KeepAlive(spool)
+			// If this succeeds, "f" is now owned by "l"
+			if err = l.Init(ctx, desc, f); err != nil {
+				return errors.Join(err, f.Close())
+			}
+			*cl = closeFunc(func() (err error) {
+				err = errors.Join(l.Close(), f.Close())
+				// Using this KeepAlive keeps the cached file descriptor live until
+				// all users of the blob have cleaned up. This should be after "f"
+				// is closed so that the cached-owned file descriptor outlives any
+				// reopened copies. There's no explicit association of these file
+				// descriptors, it's all kernel-side book-keeping.
+				runtime.KeepAlive(spool)
+				return err
+			})
+		case errUseHTTPReader: // Try to construct an [httpreader.Reader].
+			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 errors.Join(err, rd.Close())
+				}
+				*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
+				forceFetch = true
+				goto Fetch
+			default:
+				return err
+			}
+		default:
 			return err
-		})
+		}
 
 		return nil
 	}
 }
 
+// ErrUseHTTPReader is a sentinel value to signal that the calling code should
+// construct an [httpreader.Reader] rather than reopen the [os.File].
+var errUseHTTPReader = errors.New("use htttpreader.Reader")
+
 // CloseFunc is an adapter in the vein of [http.HandlerFunc].
 type closeFunc func() error
 
@@ -173,12 +223,110 @@ 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()
+	if req.Header == nil {
+		req.Header = make(http.Header)
+	}
+	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`
+	)
+	ctVal := cmp.Or(res.Header.Get(ctKey), `application/octet-stream`)
+	ct, _, err := mime.ParseMediaType(ctVal)
+	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 +358,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 +384,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 +572,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$`)