Merge pull request #1636 from tonistiigi/resolver-pool

resolver: add better pooling and custom authenticator
moby · Aug 14, 2020 · 661cafc · 661cafc
2 parents 8861f58 + e650092
commit 661cafc
Show file tree

Hide file tree

Showing 29 changed files with 1,056 additions and 367 deletions.
diff --git a/cache/remotecache/local/local.go b/cache/remotecache/local/local.go
@@ -76,7 +76,7 @@ func getContentStore(ctx context.Context, sm *session.Manager, g session.Group,
 	timeoutCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 	defer cancel()
 
-	caller, err := sm.Get(timeoutCtx, sessionID)
+	caller, err := sm.Get(timeoutCtx, sessionID, false)
 	if err != nil {
 		return nil, err
 	}

diff --git a/cache/remotecache/registry/registry.go b/cache/remotecache/registry/registry.go
@@ -37,7 +37,7 @@ func ResolveCacheExporterFunc(sm *session.Manager, hosts docker.RegistryHosts) r
 		if err != nil {
 			return nil, err
 		}
-		remote := resolver.New(hosts, resolver.NewSessionAuthenticator(sm, g))
+		remote := resolver.DefaultPool.GetResolver(hosts, ref, "push", sm, g)
 		pusher, err := remote.Pusher(ctx, ref)
 		if err != nil {
 			return nil, err
@@ -52,7 +52,7 @@ func ResolveCacheImporterFunc(sm *session.Manager, cs content.Store, hosts docke
 		if err != nil {
 			return nil, specs.Descriptor{}, err
 		}
-		remote := resolver.New(hosts, resolver.NewSessionAuthenticator(sm, g))
+		remote := resolver.DefaultPool.GetResolver(hosts, ref, "pull", sm, g)
 		xref, desc, err := remote.Resolve(ctx, ref)
 		if err != nil {
 			return nil, specs.Descriptor{}, err

diff --git a/exporter/local/export.go b/exporter/local/export.go
@@ -51,7 +51,7 @@ func (e *localExporterInstance) Export(ctx context.Context, inp exporter.Source,
 	timeoutCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
 	defer cancel()
 
-	caller, err := e.opt.SessionManager.Get(timeoutCtx, sessionID)
+	caller, err := e.opt.SessionManager.Get(timeoutCtx, sessionID, false)
 	if err != nil {
 		return nil, err
 	}

diff --git a/exporter/oci/export.go b/exporter/oci/export.go
@@ -165,7 +165,7 @@ func (e *imageExporterInstance) Export(ctx context.Context, src exporter.Source,
 	timeoutCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
 	defer cancel()
 
-	caller, err := e.opt.SessionManager.Get(timeoutCtx, sessionID)
+	caller, err := e.opt.SessionManager.Get(timeoutCtx, sessionID, false)
 	if err != nil {
 		return nil, err
 	}

diff --git a/exporter/tar/export.go b/exporter/tar/export.go
@@ -135,7 +135,7 @@ func (e *localExporterInstance) Export(ctx context.Context, inp exporter.Source,
 	timeoutCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
 	defer cancel()
 
-	caller, err := e.opt.SessionManager.Get(timeoutCtx, sessionID)
+	caller, err := e.opt.SessionManager.Get(timeoutCtx, sessionID, false)
 	if err != nil {
 		return nil, err
 	}

diff --git a/go.mod b/go.mod
@@ -12,7 +12,7 @@ require (
 	github.com/codahale/hdrhistogram v0.0.0-20160425231609-f8ad88b59a58 // indirect
 	github.com/containerd/cgroups v0.0.0-20200710171044-318312a37340 // indirect
 	github.com/containerd/console v1.0.0
-	github.com/containerd/containerd v1.4.0-beta.2.0.20200728183644-eb6354a11860
+	github.com/containerd/containerd v1.4.0-beta.2.0.20200730150746-fa1220fce33f
 	github.com/containerd/continuity v0.0.0-20200413184840-d3ef23f19fbb
 	github.com/containerd/fifo v0.0.0-20200410184934-f15a3290365b // indirect
 	github.com/containerd/go-cni v1.0.0
@@ -71,7 +71,7 @@ require (
 )
 
 replace (
-	github.com/containerd/containerd => github.com/containerd/containerd v1.4.0-beta.2.0.20200728183644-eb6354a11860
+	github.com/containerd/containerd => github.com/containerd/containerd v1.4.0-beta.2.0.20200730150746-fa1220fce33f
 	github.com/docker/docker => github.com/docker/docker v17.12.0-ce-rc1.0.20200310163718-4634ce647cf2+incompatible
 	github.com/hashicorp/go-immutable-radix => github.com/tonistiigi/go-immutable-radix v0.0.0-20170803185627-826af9ccf0fe
 	github.com/jaguilar/vt100 => github.com/tonistiigi/vt100 v0.0.0-20190402012908-ad4c4a574305

diff --git a/go.sum b/go.sum
@@ -27,8 +27,8 @@ github.com/containerd/console v0.0.0-20180822173158-c12b1e7919c1/go.mod h1:Tj/on
 github.com/containerd/console v0.0.0-20191206165004-02ecf6a7291e/go.mod h1:8Pf4gM6VEbTNRIT26AyyU7hxdQU3MvAvxVI0sc00XBE=
 github.com/containerd/console v1.0.0 h1:fU3UuQapBs+zLJu82NhR11Rif1ny2zfMMAyPJzSN5tQ=
 github.com/containerd/console v1.0.0/go.mod h1:8Pf4gM6VEbTNRIT26AyyU7hxdQU3MvAvxVI0sc00XBE=
-github.com/containerd/containerd v1.4.0-beta.2.0.20200728183644-eb6354a11860 h1:30UR1cinmvhqtKTpQWOJada+p36BgAuXf5w+aHEJOto=
-github.com/containerd/containerd v1.4.0-beta.2.0.20200728183644-eb6354a11860/go.mod h1:bC6axHOhabU15QhwfG7w5PipXdVtMXFTttgp+kVtyUA=
+github.com/containerd/containerd v1.4.0-beta.2.0.20200730150746-fa1220fce33f h1:eSl1h+oob8CkeY7TXMs6yrs6eXzgZDbqvsOM0l+EOgk=
+github.com/containerd/containerd v1.4.0-beta.2.0.20200730150746-fa1220fce33f/go.mod h1:bC6axHOhabU15QhwfG7w5PipXdVtMXFTttgp+kVtyUA=
 github.com/containerd/continuity v0.0.0-20190426062206-aaeac12a7ffc/go.mod h1:GL3xCUCBDV3CZiTSEKksMWbLE66hEyuu9qyDOOqM47Y=
 github.com/containerd/continuity v0.0.0-20200228182428-0f16d7a0959c/go.mod h1:Dq467ZllaHgAtVp4p1xUQWBrFXR9s/wyoTpG8zOJGkY=
 github.com/containerd/continuity v0.0.0-20200413184840-d3ef23f19fbb h1:nXPkFq8X1a9ycY3GYQpFNxHh3j2JgY7zDZfq2EXMIzk=

diff --git a/session/auth/auth.go b/session/auth/auth.go
@@ -8,10 +8,10 @@ import (
 	"google.golang.org/grpc/codes"
 )
 
-func CredentialsFunc(sm *session.Manager, g session.Group) func(string) (string, string, error) {
-	return func(host string) (string, string, error) {
-		var user, secret string
-		err := sm.Any(context.TODO(), g, func(ctx context.Context, _ string, c session.Caller) error {
+func CredentialsFunc(sm *session.Manager, g session.Group) func(string) (session, username, secret string, err error) {
+	return func(host string) (string, string, string, error) {
+		var sessionID, user, secret string
+		err := sm.Any(context.TODO(), g, func(ctx context.Context, id string, c session.Caller) error {
 			client := NewAuthClient(c.Conn())
 
 			resp, err := client.Credentials(ctx, &CredentialsRequest{
@@ -23,13 +23,14 @@ func CredentialsFunc(sm *session.Manager, g session.Group) func(string) (string,
 				}
 				return err
 			}
+			sessionID = id
 			user = resp.Username
 			secret = resp.Secret
 			return nil
 		})
 		if err != nil {
-			return "", "", err
+			return "", "", "", err
 		}
-		return user, secret, nil
+		return sessionID, user, secret, nil
 	}
 }
diff --git a/session/content/content_test.go b/session/content/content_test.go
@@ -63,7 +63,7 @@ func TestContentAttachable(t *testing.T) {
 	})
 
 	g.Go(func() error {
-		c, err := m.Get(ctx, s.ID())
+		c, err := m.Get(ctx, s.ID(), false)
 		if err != nil {
 			return err
 		}

diff --git a/session/filesync/filesync_test.go b/session/filesync/filesync_test.go
@@ -46,7 +46,7 @@ func TestFileSyncIncludePatterns(t *testing.T) {
 	})
 
 	g.Go(func() (reterr error) {
-		c, err := m.Get(ctx, s.ID())
+		c, err := m.Get(ctx, s.ID(), false)
 		if err != nil {
 			return err
 		}

diff --git a/session/group.go b/session/group.go
@@ -74,7 +74,7 @@ func (sm *Manager) Any(ctx context.Context, g Group, f func(context.Context, str
 
 		timeoutCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
 		defer cancel()
-		c, err := sm.Get(timeoutCtx, id)
+		c, err := sm.Get(timeoutCtx, id, false)
 		if err != nil {
 			lastErr = err
 			continue

diff --git a/session/manager.go b/session/manager.go
@@ -149,7 +149,7 @@ func (sm *Manager) handleConn(ctx context.Context, conn net.Conn, opts map[strin
 }
 
 // Get returns a session by ID
-func (sm *Manager) Get(ctx context.Context, id string) (Caller, error) {
+func (sm *Manager) Get(ctx context.Context, id string, noWait bool) (Caller, error) {
 	// session prefix is used to identify vertexes with different contexts so
 	// they would not collide, but for lookup we don't need the prefix
 	if p := strings.SplitN(id, ":", 2); len(p) == 2 && len(p[1]) > 0 {
@@ -180,14 +180,18 @@ func (sm *Manager) Get(ctx context.Context, id string) (Caller, error) {
 		}
 		var ok bool
 		c, ok = sm.sessions[id]
-		if !ok || c.closed() {
+		if (!ok || c.closed()) && !noWait {
 			sm.updateCondition.Wait()
 			continue
 		}
 		sm.mu.Unlock()
 		break
 	}
 
+	if c == nil {
+		return nil, nil
+	}
+
 	return c, nil
 }
 

diff --git a/source/containerimage/pull.go b/source/containerimage/pull.go
@@ -81,13 +81,8 @@ func (is *Source) ResolveImageConfig(ctx context.Context, ref string, opt llb.Re
 	}
 
 	res, err := is.g.Do(ctx, key, func(ctx context.Context) (interface{}, error) {
-		dgst, dt, err := imageutil.Config(ctx, ref, pull.NewResolver(g, pull.ResolverOpt{
-			Hosts:      is.RegistryHosts,
-			Auth:       resolver.NewSessionAuthenticator(sm, g),
-			ImageStore: is.ImageStore,
-			Mode:       rm,
-			Ref:        ref,
-		}), is.ContentStore, is.LeaseManager, opt.Platform)
+		res := resolver.DefaultPool.GetResolver(is.RegistryHosts, ref, "pull", sm, g).WithImageStore(is.ImageStore, rm)
+		dgst, dt, err := imageutil.Config(ctx, ref, res, is.ContentStore, is.LeaseManager, opt.Platform)
 		if err != nil {
 			return nil, err
 		}
@@ -117,28 +112,30 @@ func (is *Source) Resolve(ctx context.Context, id source.Identifier, sm *session
 		Src:          imageIdentifier.Reference,
 	}
 	p := &puller{
-		CacheAccessor: is.CacheAccessor,
-		LeaseManager:  is.LeaseManager,
-		Puller:        pullerUtil,
-		id:            imageIdentifier,
-		ResolverOpt: pull.ResolverOpt{
-			Hosts:      is.RegistryHosts,
-			Auth:       resolver.NewSessionAuthenticator(sm, nil),
-			ImageStore: is.ImageStore,
-			Mode:       imageIdentifier.ResolveMode,
-			Ref:        imageIdentifier.Reference.String(),
-		},
-		vtx: vtx,
+		CacheAccessor:  is.CacheAccessor,
+		LeaseManager:   is.LeaseManager,
+		Puller:         pullerUtil,
+		id:             imageIdentifier,
+		RegistryHosts:  is.RegistryHosts,
+		ImageStore:     is.ImageStore,
+		Mode:           imageIdentifier.ResolveMode,
+		Ref:            imageIdentifier.Reference.String(),
+		SessionManager: sm,
+		vtx:            vtx,
 	}
 	return p, nil
 }
 
 type puller struct {
-	CacheAccessor cache.Accessor
-	LeaseManager  leases.Manager
-	ResolverOpt   pull.ResolverOpt
-	id            *source.ImageIdentifier
-	vtx           solver.Vertex
+	CacheAccessor  cache.Accessor
+	LeaseManager   leases.Manager
+	RegistryHosts  docker.RegistryHosts
+	ImageStore     images.Store
+	Mode           source.ResolveMode
+	Ref            string
+	SessionManager *session.Manager
+	id             *source.ImageIdentifier
+	vtx            solver.Vertex
 
 	cacheKeyOnce     sync.Once
 	cacheKeyErr      error
@@ -169,11 +166,7 @@ func mainManifestKey(ctx context.Context, desc specs.Descriptor, platform specs.
 }
 
 func (p *puller) CacheKey(ctx context.Context, g session.Group, index int) (cacheKey string, cacheOpts solver.CacheOpts, cacheDone bool, err error) {
-	if p.Puller.Resolver == nil {
-		p.Puller.Resolver = pull.NewResolver(g, p.ResolverOpt)
-	} else {
-		p.ResolverOpt.Auth.AddSession(g)
-	}
+	p.Puller.Resolver = resolver.DefaultPool.GetResolver(p.RegistryHosts, p.Ref, "pull", p.SessionManager, g).WithImageStore(p.ImageStore, p.id.ResolveMode)
 
 	p.cacheKeyOnce.Do(func() {
 		ctx, done, err := leaseutil.WithLease(ctx, p.LeaseManager, leases.WithExpiration(5*time.Minute), leaseutil.MakeTemporary)
@@ -253,11 +246,7 @@ func (p *puller) CacheKey(ctx context.Context, g session.Group, index int) (cach
 }
 
 func (p *puller) Snapshot(ctx context.Context, g session.Group) (ir cache.ImmutableRef, err error) {
-	if p.Puller.Resolver == nil {
-		p.Puller.Resolver = pull.NewResolver(g, p.ResolverOpt)
-	} else {
-		p.ResolverOpt.Auth.AddSession(g)
-	}
+	p.Puller.Resolver = resolver.DefaultPool.GetResolver(p.RegistryHosts, p.Ref, "pull", p.SessionManager, g).WithImageStore(p.ImageStore, p.id.ResolveMode)
 
 	if len(p.manifest.Remote.Descriptors) == 0 {
 		return nil, nil

diff --git a/util/pull/pull.go b/util/pull/pull.go
@@ -96,9 +96,6 @@ func (p *Puller) PullManifests(ctx context.Context) (*PulledManifests, error) {
 		return nil, err
 	}
 
-	// workaround for gcr, authentication not supported on blob endpoints
-	EnsureManifestRequested(ctx, p.Resolver, p.ref)
-
 	platform := platforms.Only(p.Platform)
 
 	var mu sync.Mutex // images.Dispatch calls handlers in parallel