push: workaround deadlock in containerd pusher

Signed-off-by: Tonis Tiigi <tonistiigi@gmail.com>

cache/remotecache/registry/registry.go

@@ -10,6 +10,7 @@ import (
 	"github.com/moby/buildkit/cache/remotecache"
 	"github.com/moby/buildkit/session"
 	"github.com/moby/buildkit/util/contentutil"
+	"github.com/moby/buildkit/util/push"
 	"github.com/moby/buildkit/util/resolver"
 	"github.com/moby/buildkit/util/resolver/limited"
 	digest "github.com/opencontainers/go-digest"
@@ -48,7 +49,7 @@ func ResolveCacheExporterFunc(sm *session.Manager, hosts docker.RegistryHosts) r
 			ociMediatypes = b
 		}
 		remote := resolver.DefaultPool.GetResolver(hosts, ref, "push", sm, g)
-		pusher, err := remote.Pusher(ctx, ref)
+		pusher, err := push.Pusher(ctx, remote, ref)
 		if err != nil {
 			return nil, err
 		}

util/contentutil/refs.go

@@ -37,17 +37,21 @@ func IngesterFromRef(ref string) (content.Ingester, error) {
 		Client: http.DefaultClient,
 	})
 
-	pusher, err := remote.Pusher(context.TODO(), ref)
+	p, err := remote.Pusher(context.TODO(), ref)
 	if err != nil {
 		return nil, err
 	}
 
 	return &ingester{
 		locker: locker.New(),
-		pusher: pusher,
+		pusher: &pusher{p},
 	}, nil
 }
 
+type pusher struct {
+	remotes.Pusher
+}
+
 type ingester struct {
 	locker *locker.Locker
 	pusher remotes.Pusher

util/push/push.go

@@ -11,6 +11,7 @@ import (
 	"github.com/containerd/containerd/content"
 	"github.com/containerd/containerd/errdefs"
 	"github.com/containerd/containerd/images"
+	"github.com/containerd/containerd/remotes"
 	"github.com/containerd/containerd/remotes/docker"
 	"github.com/docker/distribution/reference"
 	"github.com/moby/buildkit/session"
@@ -28,6 +29,21 @@ import (
 	"github.com/sirupsen/logrus"
 )
 
+type pusher struct {
+	remotes.Pusher
+}
+
+// Pusher creates and new pusher instance for resolver
+// containerd resolver.Pusher() method is broken and should not be called directly
+// we need to wrap to mask interface detection
+func Pusher(ctx context.Context, resolver remotes.Resolver, ref string) (remotes.Pusher, error) {
+	p, err := resolver.Pusher(ctx, ref)
+	if err != nil {
+		return nil, err
+	}
+	return &pusher{Pusher: p}, nil
+}
+
 func Push(ctx context.Context, sm *session.Manager, sid string, provider content.Provider, manager content.Manager, dgst digest.Digest, ref string, insecure bool, hosts docker.RegistryHosts, byDigest bool, annotations map[digest.Digest]map[string]string) error {
 	desc := ocispecs.Descriptor{
 		Digest: dgst,
@@ -66,7 +82,7 @@ func Push(ctx context.Context, sm *session.Manager, sid string, provider content
 
 	resolver := resolver.DefaultPool.GetResolver(hosts, ref, scope, sm, session.NewGroup(sid))
 
-	pusher, err := resolver.Pusher(ctx, ref)
+	pusher, err := Pusher(ctx, resolver, ref)
 	if err != nil {
 		return err
 	}