Lazily initialize docker registry client (#328)

* Lazily attempt to initialize docker registry client

* Exit early if connection to docker daemon fails
* Fix mutex bugs
* Fix image buildinfo getting "stuck" and requiring killing of the process
* change execution order

---------

Co-authored-by: equanox <[email protected]>

Author: rdnt

bob/aggregate.go

@@ -7,6 +7,8 @@ import (
 	"path/filepath"
 	"strings"
 
+	"github.com/benchkram/bob/pkg/dockermobyutil"
+
 	"github.com/benchkram/bob/bob/bobfile"
 	"github.com/benchkram/bob/bob/bobfile/project"
 	"github.com/benchkram/bob/bob/global"
@@ -158,7 +160,6 @@ func (b *B) Aggregate() (aggregate *bobfile.Bobfile, err error) {
 		task.WithLocalstore(b.local)
 		task.WithEnvStore(b.nix.EnvStore())
 		task.WithBuildinfoStore(b.buildInfoStore)
-		task.WithDockerRegistryClient(b.dockerRegistryClient)
 
 		// a task must always-rebuild when caching is disabled
 		if !b.enableCaching {
@@ -218,6 +219,32 @@ func (b *B) Aggregate() (aggregate *bobfile.Bobfile, err error) {
 		aggregate.Project = aggregate.Dir()
 	}
 
+	var dockerRegistryClientInitialized bool
+
+	// Assure tasks are correctly initialised with a docker registry client.
+	// Only one registry client must be created and shared between tasks,
+	// this reduces the pressure on garbage collection for big repos.
+	for i, task := range aggregate.BTasks {
+		target, err := task.Target()
+		errz.Fatal(err)
+
+		if target != nil && len(target.DockerImages()) > 0 {
+			if !dockerRegistryClientInitialized {
+				b.dockerRegistryClient, err = dockermobyutil.NewRegistryClient()
+				if errors.Is(err, dockermobyutil.ErrConnectionFailed) {
+					errz.Fatal(usererror.Wrapm(err, fmt.Sprintf("task `%s` exports an image, but docker is not reachable", task.Name())))
+				}
+				errz.Fatal(err)
+
+				dockerRegistryClientInitialized = true
+			}
+
+			task.WithDockerRegistryClient(b.dockerRegistryClient)
+			// modify index on map since tasks are passed by value
+			aggregate.BTasks[i] = task
+		}
+	}
+
 	err = aggregate.Verify()
 	errz.Fatal(err)
 

bob/bob.go

@@ -81,8 +81,6 @@ func newBob(opts ...Option) *B {
 		enableCaching: true,
 		allowInsecure: false,
 		maxParallel:   runtime.NumCPU(),
-
-		dockerRegistryClient: dockermobyutil.NewRegistryClient(),
 	}
 
 	for _, opt := range opts {

bob/bobfile/bobfile.go

@@ -8,7 +8,6 @@ import (
 	"path/filepath"
 	"strings"
 
-	"github.com/benchkram/bob/pkg/dockermobyutil"
 	"github.com/benchkram/bob/pkg/nix"
 	storeclient "github.com/benchkram/bob/pkg/store-client"
 
@@ -139,9 +138,6 @@ func bobfileRead(dir string) (_ *Bobfile, err error) {
 		bobfile.RTasks = bobrun.RunMap{}
 	}
 
-	// a shared registry clients for all tasks.
-	dockerRegistryClient := dockermobyutil.NewRegistryClient()
-
 	// Assure tasks are initialized with their defaults
 	for key, task := range bobfile.BTasks {
 		task.SetDir(bobfile.dir)
@@ -153,7 +149,6 @@ func bobfileRead(dir string) (_ *Bobfile, err error) {
 		// This means switching to pointer types for most members.
 		task.SetEnv([]string{})
 		task.SetRebuildStrategy(bobtask.RebuildOnChange)
-		task.WithDockerRegistryClient(dockerRegistryClient)
 
 		// initialize docker registry for task
 		task.SetDependencies(initializeDependencies(dir, task.DependenciesDirty, bobfile))

bob/playbook/build_internal.go

@@ -130,11 +130,10 @@ func (p *Playbook) build(ctx context.Context, task *bobtask.Task) (pt *processed
 	taskSuccessFul = true
 
 	err = p.TaskCompleted(task.TaskID)
-	if err != nil {
-		if errors.Is(err, ErrFailed) {
-			return pt, err
-		}
+	if errors.Is(err, ErrFailed) {
+		return pt, err
 	}
+	errz.Log(err)
 	errz.Fatal(err)
 
 	taskStatus, err := p.TaskStatus(task.Name())

bobtask/target.go

@@ -10,12 +10,15 @@ import (
 )
 
 // Target takes care of populating the targets members correctly.
-// It returns a nil in case of a non existing target and a nil error.
+// It returns a nil in case of a non-existing target and a nil error.
 func (t *Task) Target() (empty target.Target, _ error) {
 	if t.target == nil {
 		return empty, nil
 	}
 
+	// attach docker registry client (if set) to target itself
+	t.target.WithDockerRegistryClient(t.dockerRegistryClient)
+
 	// ReadBuildInfo is dependent on the inputHash of the task.
 	// For this reason we cannot read build info on target creation,
 	// as this happens right after parsing the config.
@@ -41,8 +44,10 @@ func (t *Task) Target() (empty target.Target, _ error) {
 		return t.target, t.target.Resolve()
 	}
 
-	tt := t.target.WithExpected(&buildInfo.Target)
-	return tt, tt.Resolve()
+	// attach expected buildinfo
+	t.target.WithExpected(&buildInfo.Target)
+
+	return t.target, t.target.Resolve()
 }
 
 func (t *Task) TargetExists() bool {

bobtask/target/options.go

@@ -1,10 +1,5 @@
 package target
 
-import (
-	"github.com/benchkram/bob/bobtask/buildinfo"
-	"github.com/benchkram/bob/pkg/dockermobyutil"
-)
-
 type Option func(t *T)
 
 func WithDir(dir string) Option {
@@ -24,15 +19,3 @@ func WithDockerImages(images []string) Option {
 		t.dockerImages = images
 	}
 }
-
-func WithDockerRegistryClient(dockerRegistryClient dockermobyutil.RegistryClient) Option {
-	return func(t *T) {
-		t.dockerRegistryClient = dockerRegistryClient
-	}
-}
-
-func WithExpected(bi *buildinfo.Targets) Option {
-	return func(t *T) {
-		t.expected = bi
-	}
-}

bobtask/target/target.go

@@ -19,7 +19,7 @@ type Target interface {
 	FilesystemEntriesRaw() []string
 	FilesystemEntriesRawPlain() []string
 
-	WithExpected(*buildinfo.Targets) *T
+	WithExpected(*buildinfo.Targets)
 	DockerImages() []string
 
 	// AsInvalidFiles returns all FilesystemEntriesRaw as invalid with the specified reason
@@ -68,10 +68,6 @@ func New(opts ...Option) *T {
 		opt(t)
 	}
 
-	if t.dockerRegistryClient == nil {
-		t.dockerRegistryClient = dockermobyutil.NewRegistryClient()
-	}
-
 	return t
 }
 
@@ -109,9 +105,12 @@ func (t *T) FilesystemEntriesRawPlain() []string {
 	return append([]string{}, t.filesystemEntriesRaw...)
 }
 
-func (t *T) WithExpected(expected *buildinfo.Targets) *T {
+func (t *T) WithExpected(expected *buildinfo.Targets) {
 	t.expected = expected
-	return t
+}
+
+func (t *T) WithDockerRegistryClient(c dockermobyutil.RegistryClient) {
+	t.dockerRegistryClient = c
 }
 
 func (t *T) DockerImages() []string {

bobtask/target_parse.go

@@ -75,7 +75,6 @@ func (t *Task) parseTargets() error {
 			target.WithFilesystemEntries(filesystemEntries),
 			target.WithDockerImages(dockerImages),
 			target.WithDir(t.dir),
-			target.WithDockerRegistryClient(t.dockerRegistryClient),
 		)
 	}
 

bobtask/task.go

@@ -135,10 +135,6 @@ func Make(opts ...TaskOption) Task {
 		opt(&t)
 	}
 
-	if t.dockerRegistryClient == nil {
-		t.dockerRegistryClient = dockermobyutil.NewRegistryClient()
-	}
-
 	return t
 }
 

pkg/dockermobyutil/registry.go

@@ -18,7 +18,8 @@ import (
 )
 
 var (
-	ErrImageNotFound = fmt.Errorf("image not found")
+	ErrImageNotFound    = fmt.Errorf("image not found")
+	ErrConnectionFailed = errors.New("connection to docker daemon failed")
 )
 
 type RegistryClient interface {
@@ -45,7 +46,7 @@ type R struct {
 	mutex *sync.Mutex
 }
 
-func NewRegistryClient() RegistryClient {
+func NewRegistryClient() (RegistryClient, error) {
 	cli, err := client.NewClientWithOpts(
 		client.FromEnv,
 		client.WithAPIVersionNegotiation(),
@@ -59,14 +60,19 @@ func NewRegistryClient() RegistryClient {
 		archiveDir: os.TempDir(),
 	}
 
-	// Use a lock to supress parallel image reads on zfs.
+	// Use a lock to suppress parallel image reads on zfs.
 	info, err := r.client.Info(context.Background())
-	errz.Log(err)
+	if client.IsErrConnectionFailed(err) {
+		return nil, ErrConnectionFailed
+	} else if err != nil {
+		return nil, err
+	}
+
 	if info.Driver == "zfs" {
 		r.mutex = &sync.Mutex{}
 	}
 
-	return r
+	return r, nil
 }
 
 func (r *R) ImageExists(image string) (bool, error) {
@@ -107,17 +113,12 @@ func (r *R) ImageHash(image string) (string, error) {
 func (r *R) imageSaveToPath(image string, savedir string) (pathToArchive string, _ error) {
 	if r.mutex != nil {
 		r.mutex.Lock()
+		defer r.mutex.Unlock()
 	}
 	reader, err := r.client.ImageSave(context.Background(), []string{image})
 	if err != nil {
-		if r.mutex != nil {
-			r.mutex.Unlock()
-		}
 		return "", err
 	}
-	if r.mutex != nil {
-		r.mutex.Unlock()
-	}
 	defer reader.Close()
 
 	body, err := io.ReadAll(reader)

test/e2e/artifacts/artifacts_extraction_test.go

@@ -77,7 +77,8 @@ var _ = Describe("Test artifact creation and extraction", func() {
 var _ = Describe("Test artifact creation and extraction from docker targets", func() {
 	Context("in a fresh playground", func() {
 
-		mobyClient := dockermobyutil.NewRegistryClient()
+		mobyClient, err := dockermobyutil.NewRegistryClient()
+		Expect(err).NotTo(HaveOccurred())
 
 		It("should initialize bob playground", func() {
 			Expect(bob.CreatePlayground(bob.PlaygroundOptions{Dir: dir})).NotTo(HaveOccurred())

test/e2e/artifacts/artifacts_test.go

@@ -102,11 +102,12 @@ var _ = Describe("Test artifact and target invalidation", func() {
 	})
 })
 
-//  docker targets
+// docker targets
 var _ = Describe("Test artifact and docker-target invalidation", func() {
 	Context("in a fresh playground", func() {
 
-		mobyClient := dockermobyutil.NewRegistryClient()
+		mobyClient, err := dockermobyutil.NewRegistryClient()
+		Expect(err).NotTo(HaveOccurred())
 
 		It("should initialize bob playground", func() {
 			Expect(bob.CreatePlayground(bob.PlaygroundOptions{Dir: dir})).NotTo(HaveOccurred())

test/e2e/artifacts/nobuildinfo_test.go

@@ -104,7 +104,8 @@ var _ = Describe("Test artifact and target lifecycle without existing buildinfo"
 var _ = Describe("Test artifact and target lifecycle for docker images without existing buildinfo", func() {
 	Context("in a fresh playground", func() {
 
-		mobyClient := dockermobyutil.NewRegistryClient()
+		mobyClient, err := dockermobyutil.NewRegistryClient()
+		Expect(err).NotTo(HaveOccurred())
 
 		It("should initialize bob playground", func() {
 			Expect(bob.CreatePlayground(bob.PlaygroundOptions{Dir: dir})).NotTo(HaveOccurred())