refactor: code deduplication and simplification

cmd/compose/backend.go

@@ -0,0 +1,45 @@
+/*
+   Copyright 2020 Docker Compose CLI 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 compose
+
+import (
+	"time"
+
+	"github.com/docker/cli/cli/command"
+
+	"github.com/docker/compose/v5/pkg/api"
+	"github.com/docker/compose/v5/pkg/compose"
+)
+
+// withBackend creates a compose backend and passes it to fn.
+func withBackend(dockerCli command.Cli, opts *BackendOptions, fn func(api.Compose) error) error {
+	backend, err := compose.NewComposeService(dockerCli, opts.Options...)
+	if err != nil {
+		return err
+	}
+	return fn(backend)
+}
+
+// optionalTimeout converts an integer timeout (in seconds) into a *time.Duration.
+// If changed is false, nil is returned (no timeout was explicitly set).
+func optionalTimeout(t int, changed bool) *time.Duration {
+	if !changed {
+		return nil
+	}
+	d := time.Duration(t) * time.Second
+	return &d
+}

cmd/compose/kill.go

@@ -26,7 +26,6 @@ import (
 	"github.com/spf13/cobra"
 
 	"github.com/docker/compose/v5/pkg/api"
-	"github.com/docker/compose/v5/pkg/compose"
 	"github.com/docker/compose/v5/pkg/utils"
 )
 
@@ -63,19 +62,17 @@ func runKill(ctx context.Context, dockerCli command.Cli, backendOptions *Backend
 		return err
 	}
 
-	backend, err := compose.NewComposeService(dockerCli, backendOptions.Options...)
-	if err != nil {
+	return withBackend(dockerCli, backendOptions, func(backend api.Compose) error {
+		err := backend.Kill(ctx, name, api.KillOptions{
+			RemoveOrphans: opts.removeOrphans,
+			Project:       project,
+			Services:      services,
+			Signal:        opts.signal,
+		})
+		if errors.Is(err, api.ErrNoResources) {
+			_, _ = fmt.Fprintln(stdinfo(dockerCli), "No container to kill")
+			return nil
+		}
 		return err
-	}
-	err = backend.Kill(ctx, name, api.KillOptions{
-		RemoveOrphans: opts.removeOrphans,
-		Project:       project,
-		Services:      services,
-		Signal:        opts.signal,
 	})
-	if errors.Is(err, api.ErrNoResources) {
-		_, _ = fmt.Fprintln(stdinfo(dockerCli), "No container to kill")
-		return nil
-	}
-	return err
 }

cmd/compose/pause.go

@@ -23,7 +23,6 @@ import (
 	"github.com/spf13/cobra"
 
 	"github.com/docker/compose/v5/pkg/api"
-	"github.com/docker/compose/v5/pkg/compose"
 )
 
 type pauseOptions struct {
@@ -50,14 +49,11 @@ func runPause(ctx context.Context, dockerCli command.Cli, backendOptions *Backen
 	if err != nil {
 		return err
 	}
-
-	backend, err := compose.NewComposeService(dockerCli, backendOptions.Options...)
-	if err != nil {
-		return err
-	}
-	return backend.Pause(ctx, name, api.PauseOptions{
-		Services: services,
-		Project:  project,
+	return withBackend(dockerCli, backendOptions, func(backend api.Compose) error {
+		return backend.Pause(ctx, name, api.PauseOptions{
+			Services: services,
+			Project:  project,
+		})
 	})
 }
 
@@ -85,13 +81,10 @@ func runUnPause(ctx context.Context, dockerCli command.Cli, backendOptions *Back
 	if err != nil {
 		return err
 	}
-
-	backend, err := compose.NewComposeService(dockerCli, backendOptions.Options...)
-	if err != nil {
-		return err
-	}
-	return backend.UnPause(ctx, name, api.PauseOptions{
-		Services: services,
-		Project:  project,
+	return withBackend(dockerCli, backendOptions, func(backend api.Compose) error {
+		return backend.UnPause(ctx, name, api.PauseOptions{
+			Services: services,
+			Project:  project,
+		})
 	})
 }

cmd/compose/restart.go

@@ -18,13 +18,11 @@ package compose
 
 import (
 	"context"
-	"time"
 
 	"github.com/docker/cli/cli/command"
 	"github.com/spf13/cobra"
 
 	"github.com/docker/compose/v5/pkg/api"
-	"github.com/docker/compose/v5/pkg/compose"
 )
 
 type restartOptions struct {
@@ -69,20 +67,12 @@ func runRestart(ctx context.Context, dockerCli command.Cli, backendOptions *Back
 		}
 	}
 
-	var timeout *time.Duration
-	if opts.timeChanged {
-		timeoutValue := time.Duration(opts.timeout) * time.Second
-		timeout = &timeoutValue
-	}
-
-	backend, err := compose.NewComposeService(dockerCli, backendOptions.Options...)
-	if err != nil {
-		return err
-	}
-	return backend.Restart(ctx, name, api.RestartOptions{
-		Timeout:  timeout,
-		Services: services,
-		Project:  project,
-		NoDeps:   opts.noDeps,
+	return withBackend(dockerCli, backendOptions, func(backend api.Compose) error {
+		return backend.Restart(ctx, name, api.RestartOptions{
+			Timeout:  optionalTimeout(opts.timeout, opts.timeChanged),
+			Services: services,
+			Project:  project,
+			NoDeps:   opts.noDeps,
+		})
 	})
 }

cmd/compose/start.go

@@ -24,7 +24,6 @@ import (
 	"github.com/spf13/cobra"
 
 	"github.com/docker/compose/v5/pkg/api"
-	"github.com/docker/compose/v5/pkg/compose"
 )
 
 type startOptions struct {
@@ -58,20 +57,17 @@ func runStart(ctx context.Context, dockerCli command.Cli, backendOptions *Backen
 		return err
 	}
 
-	backend, err := compose.NewComposeService(dockerCli, backendOptions.Options...)
-	if err != nil {
-		return err
-	}
-
 	var timeout time.Duration
 	if opts.waitTimeout > 0 {
 		timeout = time.Duration(opts.waitTimeout) * time.Second
 	}
-	return backend.Start(ctx, name, api.StartOptions{
-		AttachTo:    services,
-		Project:     project,
-		Services:    services,
-		Wait:        opts.wait,
-		WaitTimeout: timeout,
+	return withBackend(dockerCli, backendOptions, func(backend api.Compose) error {
+		return backend.Start(ctx, name, api.StartOptions{
+			AttachTo:    services,
+			Project:     project,
+			Services:    services,
+			Wait:        opts.wait,
+			WaitTimeout: timeout,
+		})
 	})
 }

cmd/compose/stop.go

@@ -18,13 +18,11 @@ package compose
 
 import (
 	"context"
-	"time"
 
 	"github.com/docker/cli/cli/command"
 	"github.com/spf13/cobra"
 
 	"github.com/docker/compose/v5/pkg/api"
-	"github.com/docker/compose/v5/pkg/compose"
 )
 
 type stopOptions struct {
@@ -59,19 +57,11 @@ func runStop(ctx context.Context, dockerCli command.Cli, backendOptions *Backend
 	if err != nil {
 		return err
 	}
-
-	var timeout *time.Duration
-	if opts.timeChanged {
-		timeoutValue := time.Duration(opts.timeout) * time.Second
-		timeout = &timeoutValue
-	}
-	backend, err := compose.NewComposeService(dockerCli, backendOptions.Options...)
-	if err != nil {
-		return err
-	}
-	return backend.Stop(ctx, name, api.StopOptions{
-		Timeout:  timeout,
-		Services: services,
-		Project:  project,
+	return withBackend(dockerCli, backendOptions, func(backend api.Compose) error {
+		return backend.Stop(ctx, name, api.StopOptions{
+			Timeout:  optionalTimeout(opts.timeout, opts.timeChanged),
+			Services: services,
+			Project:  project,
+		})
 	})
 }

pkg/compose/containers.go

@@ -26,6 +26,7 @@ import (
 	"github.com/compose-spec/compose-go/v2/types"
 	"github.com/moby/moby/api/types/container"
 	"github.com/moby/moby/client"
+	"golang.org/x/sync/errgroup"
 
 	"github.com/docker/compose/v5/pkg/api"
 )
@@ -61,7 +62,7 @@ func getDefaultFilters(projectName string, oneOff oneOff, selectedServices ...st
 	if len(selectedServices) == 1 {
 		f.Add("label", serviceFilter(selectedServices[0]))
 	}
-	f.Add("label", hasConfigHashLabel())
+	f.Add("label", api.ConfigHashLabel)
 	switch oneOff {
 	case oneOffOnly:
 		f.Add("label", oneOffFilter(true))
@@ -166,12 +167,18 @@ func (containers Containers) names() []string {
 	return names
 }
 
-func (containers Containers) forEach(fn func(container.Summary)) {
-	for _, c := range containers {
-		fn(c)
+// forEachContainerConcurrent runs fn for every container concurrently and waits for all goroutines.
+func forEachContainerConcurrent(ctx context.Context, containers Containers, fn func(context.Context, container.Summary) error) error {
+	eg, ctx := errgroup.WithContext(ctx)
+	for _, ctr := range containers {
+		eg.Go(func() error {
+			return fn(ctx, ctr)
+		})
 	}
+	return eg.Wait()
 }
 
+// sorted sorts containers in place by canonical name and returns the (same) slice.
 func (containers Containers) sorted() Containers {
 	sort.Slice(containers, func(i, j int) bool {
 		return getCanonicalContainerName(containers[i]) < getCanonicalContainerName(containers[j])

pkg/compose/convergence.go

@@ -193,7 +193,7 @@ func (c *convergence) ensureService(ctx context.Context, project *types.Project,
 		name := getContainerProgressName(ctr)
 		switch ctr.State {
 		case container.StateRunning:
-			c.compose.events.On(runningEvent(name))
+			c.compose.events.On(newEvent(name, api.Done, api.StatusRunning))
 		case container.StateCreated:
 		case container.StateRestarting:
 		case container.StateExited:
@@ -303,34 +303,24 @@ func (c *convergence) resolveVolumeFrom(service *types.ServiceConfig) error {
 }
 
 func (c *convergence) resolveSharedNamespaces(service *types.ServiceConfig) error {
-	str := service.NetworkMode
-	if name := getDependentServiceFromMode(str); name != "" {
-		dependencies := c.getObservedState(name)
-		if len(dependencies) == 0 {
-			return fmt.Errorf("cannot share network namespace with service %s: container missing", name)
+	resolve := func(field *string, noun string) error {
+		if name := getDependentServiceFromMode(*field); name != "" {
+			dependencies := c.getObservedState(name)
+			if len(dependencies) == 0 {
+				return fmt.Errorf("cannot share %s namespace with service %s: container missing", noun, name)
+			}
+			*field = types.ContainerPrefix + dependencies.sorted()[0].ID
 		}
-		service.NetworkMode = types.ContainerPrefix + dependencies.sorted()[0].ID
+		return nil
 	}
 
-	str = service.Ipc
-	if name := getDependentServiceFromMode(str); name != "" {
-		dependencies := c.getObservedState(name)
-		if len(dependencies) == 0 {
-			return fmt.Errorf("cannot share IPC namespace with service %s: container missing", name)
-		}
-		service.Ipc = types.ContainerPrefix + dependencies.sorted()[0].ID
+	if err := resolve(&service.NetworkMode, "network"); err != nil {
+		return err
 	}
-
-	str = service.Pid
-	if name := getDependentServiceFromMode(str); name != "" {
-		dependencies := c.getObservedState(name)
-		if len(dependencies) == 0 {
-			return fmt.Errorf("cannot share PID namespace with service %s: container missing", name)
-		}
-		service.Pid = types.ContainerPrefix + dependencies.sorted()[0].ID
+	if err := resolve(&service.Ipc, "IPC"); err != nil {
+		return err
 	}
-
-	return nil
+	return resolve(&service.Pid, "PID")
 }
 
 func (c *convergence) mustRecreate(expected types.ServiceConfig, actual container.Summary, policy string) (bool, error) {
@@ -927,7 +917,7 @@ func (s *composeService) startService(ctx context.Context,
 		}
 
 		eventName := getContainerProgressName(ctr)
-		s.events.On(startingEvent(eventName))
+		s.events.On(newEvent(eventName, api.Working, api.StatusStarting))
 		_, err = s.apiClient().ContainerStart(ctx, ctr.ID, client.ContainerStartOptions{})
 		if err != nil {
 			return err
@@ -940,7 +930,7 @@ func (s *composeService) startService(ctx context.Context,
 			}
 		}
 
-		s.events.On(startedEvent(eventName))
+		s.events.On(newEvent(eventName, api.Done, api.StatusStarted))
 	}
 	return nil
 }

pkg/compose/convergence_test.go

@@ -73,7 +73,7 @@ func TestServiceLinks(t *testing.T) {
 		Filters: projectFilter(testProject).Add("label",
 			serviceFilter("db"),
 			oneOffFilter(false),
-			hasConfigHashLabel(),
+			api.ConfigHashLabel,
 		),
 		All: true,
 	}
@@ -201,7 +201,7 @@ func TestServiceLinks(t *testing.T) {
 			Filters: projectFilter(testProject).Add("label",
 				serviceFilter("web"),
 				oneOffFilter(false),
-				hasConfigHashLabel(),
+				api.ConfigHashLabel,
 			),
 			All: true,
 		}