Repository: oras-project/oras
Pull request #464 (2022-07-29 07:12:04)

perf: add FetchReference support in the caching proxy
This PR adds FetchReference() for proxied cache, which will be used for optimizing fetching manifest content (when `--descriptor` is not provided) after `oras.Resolve()`.

Related to #301

Signed-off-by: Billy Zha <jinzha1@microsoft.com>

codecov-commenter: # [Codecov](https://codecov.io/gh/oras-project/oras/pull/464?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=oras-project) Report
> Merging [#464](https://codecov.io/gh/oras-project/oras/pull/464?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=oras-project) (9a67b1d) into [main](https://codecov.io/gh/oras-project/oras/commit/96900e7cfb819fc2293fd8a7e5ca6a058392fdcf?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=oras-project) (96900e7) will **increase** coverage by `1.49%`.
> The diff coverage is `62.90%`.

> :exclamation: Current head 9a67b1d differs from pull request most recent head c43df2d. Consider uploading reports for the commit c43df2d to get more accurate results

```diff
@@            Coverage Diff             @@
##             main     #464      +/-   ##
==========================================
+ Coverage   55.69%   57.19%   +1.49%     
==========================================
  Files           6        7       +1     
  Lines         237      299      +62     
==========================================
+ Hits          132      171      +39     
- Misses         90      110      +20     
- Partials       15       18       +3     
```


| [Impacted Files](https://codecov.io/gh/oras-project/oras/pull/464?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=oras-project) | Coverage Δ | |
|---|---|---|
| [internal/cache/proxy.go](https://codecov.io/gh/oras-project/oras/pull/464/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=oras-project#diff-aW50ZXJuYWwvY2FjaGUvcHJveHkuZ28=) | `62.90% <62.90%> (ø)` | |

Help us with your feedback. Take ten seconds to tell us [how you rate us](https://about.codecov.io/nps?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=oras-project).


Patch:
diff --git a/go.mod b/go.mod
index 52e8185..62a3d44 100644
--- a/go.mod
+++ b/go.mod
@@ -12,7 +12,7 @@ require (
 	github.com/sirupsen/logrus v1.9.0
 	github.com/spf13/cobra v1.5.0
 	github.com/spf13/pflag v1.0.5
-	oras.land/oras-go/v2 v2.0.0-20220726123138-064752f3088e
+	oras.land/oras-go/v2 v2.0.0-20220804053649-e135557babfa
 )
 
 require (
diff --git a/go.sum b/go.sum
index f1051c1..e031b60 100644
--- a/go.sum
+++ b/go.sum
@@ -67,5 +67,5 @@ gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c h1:dUUwHk2QECo/6vqA44rthZ8ie
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gotest.tools/v3 v3.0.2 h1:kG1BFyqVHuQoVQiR1bWGnfz/fmHvvuiSPIV7rvl360E=
 gotest.tools/v3 v3.0.2/go.mod h1:3SzNCllyD9/Y+b5r9JIKQ474KzkZyqLqEfYqMsX94Bk=
-oras.land/oras-go/v2 v2.0.0-20220726123138-064752f3088e h1:0j9CmQd4yoAwwAeykCWKf1i/X45yvOgER9XZNtDGI7w=
-oras.land/oras-go/v2 v2.0.0-20220726123138-064752f3088e/go.mod h1:IZRIoIJqkAH6x0pL3tVnpyPUyZgthjSyPcH2kgJvBMo=
+oras.land/oras-go/v2 v2.0.0-20220804053649-e135557babfa h1:/K4LND5cuAwZTGZ793qPQOppK8+vNX1JvSFNIEcUDKM=
+oras.land/oras-go/v2 v2.0.0-20220804053649-e135557babfa/go.mod h1:IZRIoIJqkAH6x0pL3tVnpyPUyZgthjSyPcH2kgJvBMo=
diff --git a/internal/cache/target.go b/internal/cache/target.go
index b7e8b7b..2955233 100644
--- a/internal/cache/target.go
+++ b/internal/cache/target.go
@@ -23,6 +23,7 @@ import (
 	ocispec "github.com/opencontainers/image-spec/specs-go/v1"
 	"oras.land/oras-go/v2"
 	"oras.land/oras-go/v2/content"
+	"oras.land/oras-go/v2/registry"
 )
 
 type closer func() error
@@ -39,57 +40,100 @@ type target struct {
 
 // New generates a new target storage with caching.
 func New(source oras.Target, cache content.Storage) oras.Target {
-	return &target{
+	t := &target{
 		Target: source,
 		cache:  cache,
 	}
+	if refFetcher, ok := source.(registry.ReferenceFetcher); ok {
+		return &referenceTarget{
+			target:           t,
+			ReferenceFetcher: refFetcher,
+		}
+	}
+	return t
 }
 
 // Fetch fetches the content identified by the descriptor.
-func (p *target) Fetch(ctx context.Context, target ocispec.Descriptor) (io.ReadCloser, error) {
-	rc, err := p.cache.Fetch(ctx, target)
+func (t *target) Fetch(ctx context.Context, target ocispec.Descriptor) (io.ReadCloser, error) {
+	rc, err := t.cache.Fetch(ctx, target)
 	if err == nil {
+		// Fetch from cache
 		return rc, nil
 	}
 
-	rc, err = p.Target.Fetch(ctx, target)
-	if err != nil {
+	if rc, err = t.Target.Fetch(ctx, target); err != nil {
 		return nil, err
 	}
+
+	// Fetch from origin with caching
+	return t.cacheReadCloser(ctx, rc, target), nil
+}
+
+func (t *target) cacheReadCloser(ctx context.Context, rc io.ReadCloser, target ocispec.Descriptor) io.ReadCloser {
 	pr, pw := io.Pipe()
 	var wg sync.WaitGroup
+
 	wg.Add(1)
 	var pushErr error
 	go func() {
 		defer wg.Done()
-		pushErr = p.cache.Push(ctx, target, pr)
+		pushErr = t.cache.Push(ctx, target, pr)
 	}()
-	c := closer(func() error {
-		rcErr := rc.Close()
-		if err := pw.Close(); err != nil {
-			return err
-		}
-		wg.Wait()
-		if pushErr != nil {
-			return pushErr
-		}
-		return rcErr
-	})
 
 	return struct {
 		io.Reader
 		io.Closer
 	}{
 		Reader: io.TeeReader(rc, pw),
-		Closer: c,
-	}, nil
+		Closer: closer(func() error {
+			rcErr := rc.Close()
+			if err := pw.Close(); err != nil {
+				return err
+			}
+			wg.Wait()
+			if pushErr != nil {
+				return pushErr
+			}
+			return rcErr
+		}),
+	}
 }
 
 // Exists returns true if the described content exists.
-func (p *target) Exists(ctx context.Context, desc ocispec.Descriptor) (bool, error) {
-	exists, err := p.cache.Exists(ctx, desc)
+func (t *target) Exists(ctx context.Context, desc ocispec.Descriptor) (bool, error) {
+	exists, err := t.cache.Exists(ctx, desc)
 	if err == nil && exists {
 		return true, nil
 	}
-	return p.Target.Exists(ctx, desc)
+	return t.Target.Exists(ctx, desc)
+}
+
+// Cache referenceTarget struct.
+type referenceTarget struct {
+	*target
+	registry.ReferenceFetcher
+}
+
+// FetchReference fetches the content identified by the reference from the
+// remote and cache the fetched content.
+// Cached content will only be read via Fetch, FetchReference will always fetch
+// From origin.
+func (t *referenceTarget) FetchReference(ctx context.Context, reference string) (ocispec.Descriptor, io.ReadCloser, error) {
+	target, rc, err := t.ReferenceFetcher.FetchReference(ctx, reference)
+	if err != nil {
+		return ocispec.Descriptor{}, nil, err
+	}
+
+	// skip caching if the content already exists in cache
+	exists, err := t.cache.Exists(ctx, target)
+	if err != nil {
+		return ocispec.Descriptor{}, nil, err
+	}
+	if exists {
+		// no need to do tee'd push
+		return target, rc, nil
+	}
+
+	// Fetch from origin with caching
+	return target, t.cacheReadCloser(ctx, rc, target), nil
 }


Test patch:
diff --git a/internal/cache/target_test.go b/internal/cache/target_test.go
new file mode 100644
index 0000000..bc44855
--- /dev/null
+++ b/internal/cache/target_test.go
@@ -0,0 +1,223 @@
+/*
+Copyright The ORAS Authors.
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package cache
+
+import (
+	"bytes"
+	"context"
+	_ "crypto/sha256"
+	"fmt"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"reflect"
+	"strconv"
+	"sync/atomic"
+	"testing"
+
+	"github.com/opencontainers/go-digest"
+	ocispec "github.com/opencontainers/image-spec/specs-go/v1"
+	"oras.land/oras-go/v2/content"
+	"oras.land/oras-go/v2/content/memory"
+	"oras.land/oras-go/v2/registry"
+	"oras.land/oras-go/v2/registry/remote"
+)
+
+func TestProxy_fetchCache(t *testing.T) {
+	blob := []byte("hello world")
+	desc := ocispec.Descriptor{
+		MediaType: "test",
+		Digest:    digest.FromBytes(blob),
+		Size:      int64(len(blob)),
+	}
+
+	p := New(memory.New(), memory.New())
+	ctx := context.Background()
+
+	err := p.Push(ctx, desc, bytes.NewReader(blob))
+	if err != nil {
+		t.Fatal("Proxy.Push() error =", err)
+	}
+
+	// first fetch
+	exists, err := p.Exists(ctx, desc)
+	if err != nil {
+		t.Fatal("Proxy.Exists() error =", err)
+	}
+	if !exists {
+		t.Errorf("Proxy.Exists() = %v, want %v", exists, true)
+	}
+	got, err := content.FetchAll(ctx, p, desc)
+	if err != nil {
+		t.Fatal("Proxy.Fetch() error =", err)
+	}
+	if !bytes.Equal(got, blob) {
+		t.Errorf("Proxy.Fetch() = %v, want %v", got, blob)
+	}
+
+	// repeated fetch should not touch base CAS
+	// nil base will generate panic if the base CAS is touched
+	p.(*target).Target = nil
+
+	exists, err = p.Exists(ctx, desc)
+	if err != nil {
+		t.Fatal("Proxy.Exists() error =", err)
+	}
+	if !exists {
+		t.Errorf("Proxy.Exists() = %v, want %v", exists, true)
+	}
+	got, err = content.FetchAll(ctx, p, desc)
+	if err != nil {
+		t.Fatal("Proxy.Fetch() error =", err)
+	}
+	if !bytes.Equal(got, blob) {
+		t.Errorf("Proxy.Fetch() = %v, want %v", got, blob)
+	}
+}
+
+func TestProxy_pushPassThrough(t *testing.T) {
+	blob := []byte("hello world")
+	desc := ocispec.Descriptor{
+		MediaType: "test",
+		Digest:    digest.FromBytes(blob),
+		Size:      int64(len(blob)),
+	}
+
+	p := New(memory.New(), memory.New())
+	ctx := context.Background()
+
+	// before push
+	exists, err := p.Exists(ctx, desc)
+	if err != nil {
+		t.Fatal("Proxy.Exists() error =", err)
+	}
+	if exists {
+		t.Errorf("Proxy.Exists() = %v, want %v", exists, false)
+	}
+
+	// push
+	err = p.Push(ctx, desc, bytes.NewReader(blob))
+	if err != nil {
+		t.Fatal("Proxy.Push() error =", err)
+	}
+
+	// after push
+	exists, err = p.Exists(ctx, desc)
+	if err != nil {
+		t.Fatal("Proxy.Exists() error =", err)
+	}
+	if !exists {
+		t.Errorf("Proxy.Exists() = %v, want %v", exists, true)
+	}
+	got, err := content.FetchAll(ctx, p, desc)
+	if err != nil {
+		t.Fatal("Proxy.Fetch() error =", err)
+	}
+	if !bytes.Equal(got, blob) {
+		t.Errorf("Proxy.Fetch() = %v, want %v", got, blob)
+	}
+}
+
+func TestProxy_fetchReference(t *testing.T) {
+	// mocked variables
+	blob := []byte("{}")
+	repoName := "test/repo"
+	tagName := "test-tag"
+	mediaType := ocispec.MediaTypeImageManifest
+	digest := digest.FromBytes(blob)
+	desc := ocispec.Descriptor{
+		MediaType: mediaType,
+		Digest:    digest,
+		Size:      int64(len(blob)),
+	}
+
+	// mocked remote registry
+	var requestCount, wantRequestCount int64
+	var successCount, wantSuccessCount int64
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		atomic.AddInt64(&requestCount, 1)
+
+		if r.Method == http.MethodGet &&
+			(r.URL.Path == fmt.Sprintf("/v2/%s/manifests/%s", repoName, tagName) ||
+				r.URL.Path == fmt.Sprintf("/v2/%s/manifests/%s", repoName, digest)) {
+			w.Header().Set("Content-Type", mediaType)
+			w.Header().Set("Docker-Content-Digest", digest.String())
+			w.Header().Set("Content-Length", strconv.Itoa(len([]byte(blob))))
+			w.WriteHeader(http.StatusOK)
+			w.Write(blob)
+			atomic.AddInt64(&successCount, 1)
+			return
+		}
+		t.Errorf("unexpected access: %s %s", r.Method, r.URL)
+		w.WriteHeader(http.StatusBadRequest)
+	}))
+	defer ts.Close()
+	uri, err := url.Parse(ts.URL)
+	if err != nil {
+		t.Fatalf("invalid test http server: %v", err)
+	}
+	repo, err := remote.NewRepository(fmt.Sprintf("%s/%s:%s", uri.Host, repoName, tagName))
+	if err != nil {
+		t.Fatalf("NewRepository() error = %v", err)
+	}
+	repo.PlainHTTP = true
+	p := New(repo, memory.New())
+	ctx := context.Background()
+
+	// first fetch reference
+	gotDesc, rc, err := p.(registry.ReferenceFetcher).FetchReference(ctx, repo.Reference.Reference)
+	if err != nil {
+		t.Fatal("ReferenceTarget.FetchReference() error =", err)
+	}
+	if !reflect.DeepEqual(gotDesc, desc) {
+		t.Fatalf("ReferenceTarget.FetchReference() got %v, want %v", gotDesc, desc)
+	}
+	got, err := io.ReadAll(rc)
+	if err != nil {
+		t.Fatal("io.ReadAll() error =", err)
+	}
+	err = rc.Close()
+	if err != nil {
+		t.Error("ReferenceTarget.FetchReference().Close() error =", err)
+	}
+
+	if !bytes.Equal(got, blob) {
+		t.Errorf("ReferenceTarget.Fetch() = %v, want %v", got, blob)
+	}
+	if wantRequestCount++; requestCount != wantRequestCount {
+		t.Errorf("unexpected number of requests: %d, want %d", requestCount, wantRequestCount)
+	}
+	if wantSuccessCount++; successCount != wantSuccessCount {
+		t.Errorf("unexpected number of successful requests: %d, want %d", successCount, wantSuccessCount)
+	}
+
+	// repeated fetch should not touch base CAS
+	p.(*referenceTarget).Target = nil
+	got, err = content.FetchAll(ctx, p, desc)
+	if err != nil {
+		t.Fatal("ReferenceTarget.Fetch() error =", err)
+	}
+	if !bytes.Equal(got, blob) {
+		t.Errorf("ReferenceTarget.Fetch() = %v, want %v", got, blob)
+	}
+	if requestCount != wantRequestCount {
+		t.Errorf("unexpected number of requests: %d, want %d", requestCount, wantRequestCount)
+	}
+	if successCount != wantSuccessCount {
+		t.Errorf("unexpected number of successful requests: %d, want %d", successCount, wantSuccessCount)
+	}
+}