[registry-facade] Parallelise IPFS CID lookup

Author: csweichel

components/registry-facade/pkg/registry/blob.go

@@ -85,7 +85,7 @@ type blobHandler struct {
 	Spec              *api.ImageSpec
 	Resolver          remotes.Resolver
 	Store             content.Store
-	IPFS              *IPFSStore
+	IPFS              *IPFSBlobCache
 	AdditionalSources []BlobSource
 	ConfigModifier    ConfigModifier
 

components/registry-facade/pkg/registry/ipfs.go

@@ -7,6 +7,7 @@ package registry
 import (
 	"context"
 	"io"
+	"sync"
 	"time"
 
 	"github.com/gitpod-io/gitpod/common-go/log"
@@ -18,30 +19,42 @@ import (
 	ociv1 "github.com/opencontainers/image-spec/specs-go/v1"
 )
 
+// ipfsManifestModifier modifies a manifest and adds IPFS URLs to the layers
 func (reg *Registry) ipfsManifestModifier(mf *ociv1.Manifest) error {
 	if reg.IPFS == nil {
 		return nil
 	}
 
 	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 	defer cancel()
+
+	var wg sync.WaitGroup
 	for i, l := range mf.Layers {
-		url, _ := reg.IPFS.Get(ctx, l.Digest)
-		if url == "" {
-			continue
-		}
-		mf.Layers[i].URLs = append(mf.Layers[i].URLs, url)
+		wg.Add(1)
+		go func(i int, dgst digest.Digest) {
+			defer wg.Done()
+
+			url, _ := reg.IPFS.Get(ctx, dgst)
+			if url == "" {
+				return
+			}
+			mf.Layers[i].URLs = append(mf.Layers[i].URLs, url)
+		}(i, l.Digest)
 	}
+	wg.Wait()
 
 	return nil
 }
 
-type IPFSStore struct {
+// IPFSBlobCache can cache blobs in IPFS
+type IPFSBlobCache struct {
 	Redis *redis.Client
 	IPFS  ipfs.CoreAPI
 }
 
-func (store *IPFSStore) Get(ctx context.Context, dgst digest.Digest) (ipfsURL string, err error) {
+// Get retrieves the IPFS URL for a previously stored blob.
+// Returns an error if the blob is not stored in IPFS yet.
+func (store *IPFSBlobCache) Get(ctx context.Context, dgst digest.Digest) (ipfsURL string, err error) {
 	if store == nil || store.IPFS == nil || store.Redis == nil {
 		return "", nil
 	}
@@ -54,20 +67,8 @@ func (store *IPFSStore) Get(ctx context.Context, dgst digest.Digest) (ipfsURL st
 	return "ipfs://" + res, nil
 }
 
-func (store *IPFSStore) Has(ctx context.Context, dgst digest.Digest) (ok bool, err error) {
-	if store == nil || store.IPFS == nil || store.Redis == nil {
-		return false, nil
-	}
-
-	res := store.Redis.Exists(ctx, dgst.String())
-	if err := res.Err(); err != nil {
-		return false, err
-	}
-
-	return res.Val() == 1, nil
-}
-
-func (store *IPFSStore) Store(ctx context.Context, dgst digest.Digest, content io.Reader) (err error) {
+// Store stores a blob in IPFS. Will happily overwrite/re-upload a blob.
+func (store *IPFSBlobCache) Store(ctx context.Context, dgst digest.Digest, content io.Reader) (err error) {
 	if store == nil || store.IPFS == nil || store.Redis == nil {
 		return nil
 	}

components/registry-facade/pkg/registry/registry.go

@@ -70,7 +70,7 @@ type Registry struct {
 	Config         config.Config
 	Resolver       ResolverProvider
 	Store          content.Store
-	IPFS           *IPFSStore
+	IPFS           *IPFSBlobCache
 	LayerSource    LayerSource
 	ConfigModifier ConfigModifier
 	SpecProvider   map[string]ImageSpecProvider
@@ -190,7 +190,7 @@ func NewRegistry(cfg config.Config, newResolver ResolverProvider, reg prometheus
 		specProvider[api.ProviderPrefixFixed] = FixedImageSpecProvider(fp)
 	}
 
-	var ipfs *IPFSStore
+	var ipfs *IPFSBlobCache
 	if cfg.IPFSCache != nil && cfg.IPFSCache.Enabled {
 		addr := cfg.IPFSCache.IPFSAddr
 		if ipfsHost := os.Getenv("IPFS_HOST"); ipfsHost != "" {
@@ -210,7 +210,7 @@ func NewRegistry(cfg config.Config, newResolver ResolverProvider, reg prometheus
 			return nil, xerrors.Errorf("cannot connect to Redis: %w", err)
 		}
 
-		ipfs = &IPFSStore{
+		ipfs = &IPFSBlobCache{
 			Redis: rdc,
 			IPFS:  core,
 		}