Handle service scale with container numbering

Signed-off-by: Nicolas De Loof <[email protected]>

local/backend.go

@@ -73,3 +73,4 @@ func (s *local) ResourceService() resources.Service {
 }
 
 
+

local/compose.go

@@ -22,8 +22,10 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"golang.org/x/sync/errgroup"
 	"io"
 	"path/filepath"
+	"strconv"
 	"strings"
 	"sync"
 
@@ -169,13 +171,42 @@ func (s *local) Down(ctx context.Context, projectName string) error {
 	if err != nil {
 		return err
 	}
+
+	eg, ctx := errgroup.WithContext(ctx)
+	w := progress.ContextWriter(ctx)
 	for _, c := range list {
-		err := s.containerService.Stop(ctx, c.ID, nil)
-		if err != nil {
-			return err
-		}
+		container := c
+		eg.Go(func() error {
+			w.Event(progress.Event{
+				ID:     getContainerName(container),
+				Text:   "Stopping",
+				Status: progress.Working,
+				Done:   false,
+			})
+			err := s.containerService.Stop(ctx, container.ID, nil)
+			if err != nil {
+				return err
+			}
+			w.Event(progress.Event{
+				ID:     getContainerName(container),
+				Text:   "Removing",
+				Status: progress.Working,
+				Done:   false,
+			})
+			err = s.containerService.Delete(ctx, container.ID, containers.DeleteRequest{})
+			if err != nil {
+				return err
+			}
+			w.Event(progress.Event{
+				ID:     getContainerName(container),
+				Text:   "Removed",
+				Status: progress.Done,
+				Done:   true,
+			})
+			return nil
+		})
 	}
-	return nil
+	return eg.Wait()
 }
 
 func (s *local) Logs(ctx context.Context, projectName string, w io.Writer) error {
@@ -250,7 +281,7 @@ func (s *local) Convert(ctx context.Context, project *types.Project, format stri
 	}
 }
 
-func getContainerCreateOptions(p *types.Project, s types.ServiceConfig, inherit *moby.Container) (*container.Config, *container.HostConfig, *network.NetworkingConfig, error) {
+func getContainerCreateOptions(p *types.Project, s types.ServiceConfig, number int, inherit *moby.Container) (*container.Config, *container.HostConfig, *network.NetworkingConfig, error) {
 	hash, err := jsonHash(s)
 	if err != nil {
 		return nil, nil, nil, err
@@ -259,6 +290,7 @@ func getContainerCreateOptions(p *types.Project, s types.ServiceConfig, inherit
 		"com.docker.compose.project":     p.Name,
 		"com.docker.compose.service":     s.Name,
 		"com.docker.compose.config-hash": hash,
+		"com.docker.compose.container-number": strconv.Itoa(number),
 	}
 
 	var (
@@ -523,7 +555,7 @@ func (s *local) ensureNetwork(ctx context.Context, n types.NetworkConfig) error
 			}
 			w.Event(progress.Event{
 				ID:         fmt.Sprintf("Network %q", n.Name),
-				Status:     progress.Working,
+				Status:     progress.Done,
 				StatusText: "Created",
 				Done:       true,
 			})

local/convergence.go

@@ -21,14 +21,14 @@ package local
 import (
 	"context"
 	"fmt"
-	"github.com/docker/docker/api/types/network"
-
+	"github.com/compose-spec/compose-go/types"
 	"github.com/docker/compose-cli/api/containers"
 	"github.com/docker/compose-cli/progress"
-
-	"github.com/compose-spec/compose-go/types"
 	moby "github.com/docker/docker/api/types"
 	"github.com/docker/docker/api/types/filters"
+	"github.com/docker/docker/api/types/network"
+	"golang.org/x/sync/errgroup"
+	"strconv"
 )
 
 func (s *local) ensureService(ctx context.Context, project *types.Project, service types.ServiceConfig) error {
@@ -42,30 +42,90 @@ func (s *local) ensureService(ctx context.Context, project *types.Project, servi
 		return err
 	}
 
+	scale := getScale(service)
+
+	eg, ctx := errgroup.WithContext(ctx)
+	if len(actual) < scale {
+		next, err := nextContainerNumber(actual)
+		if err != nil {
+			return err
+		}
+		missing := scale - len(actual)
+		for i := 0; i < missing; i++ {
+			number := next + i
+			name := fmt.Sprintf("%s_%s_%d", project.Name, service.Name, number)
+			eg.Go(func() error {
+				return s.createContainer(ctx, project, service, name, number)
+			})
+		}
+	}
+
+	if len(actual) > scale {
+		for i := scale; i < len(actual); i++ {
+			container := actual[i]
+			eg.Go(func() error {
+				err := s.containerService.Stop(ctx, container.ID, nil)
+				if err != nil {
+					return err
+				}
+				return s.containerService.Delete(ctx, container.ID, containers.DeleteRequest{})
+			})
+		}
+		actual = actual[:scale]
+	}
+
 	expected, err := jsonHash(service)
 	if err != nil {
 		return err
 	}
+	for _, container := range actual {
+		container := container
+		diverged := container.Labels["com.docker.compose.config-hash"] != expected
+		if diverged {
+			eg.Go(func() error {
+				return s.recreateContainer(ctx, project, service, container)
+			})
+			continue
+		}
 
-	if len(actual) == 0 {
-		return s.createContainer(ctx, project, service)
-	}
+		if container.State == "running" {
+			// already running, skip
+			continue
+		}
 
-	container := actual[0] // TODO handle services with replicas
-	diverged := container.Labels["com.docker.compose.config-hash"] != expected
-	if diverged {
-		return s.recreateContainer(ctx, project, service, container)
+		eg.Go(func() error {
+			return s.restartContainer(ctx, service, container)
+		})
 	}
+	return eg.Wait()
+}
 
-	if container.State == "running" {
-		// already running, skip
-		return nil
+func nextContainerNumber(containers []moby.Container) (int, error) {
+	max := 0
+	for _, c := range containers {
+		n, err := strconv.Atoi(c.Labels["com.docker.compose.container-number"])
+		if err != nil {
+			return 0, err
+		}
+		if n > max {
+			max = n
+		}
 	}
+	return max + 1, nil
 
-	return s.restartContainer(ctx, service, container)
 }
 
-func (s *local) createContainer(ctx context.Context, project *types.Project, service types.ServiceConfig) error {
+func getScale(config types.ServiceConfig) int {
+	if config.Deploy != nil && config.Deploy.Replicas != nil {
+		return int(*config.Deploy.Replicas)
+	}
+	if config.Scale != 0 {
+		return config.Scale
+	}
+	return 1
+}
+
+func (s *local) createContainer(ctx context.Context, project *types.Project, service types.ServiceConfig, name string, number int) error {
 	w := progress.ContextWriter(ctx)
 	w.Event(progress.Event{
 		ID:         fmt.Sprintf("Service %q", service.Name),
@@ -73,8 +133,7 @@ func (s *local) createContainer(ctx context.Context, project *types.Project, ser
 		StatusText: "Create",
 		Done:       false,
 	})
-	name := fmt.Sprintf("%s_%s", project.Name, service.Name)
-	err := s.runContainer(ctx, project, service, name, nil)
+	err := s.runContainer(ctx, project, service, name, number, nil)
 	if err != nil {
 		return err
 	}
@@ -105,7 +164,11 @@ func (s *local) recreateContainer(ctx context.Context, project *types.Project, s
 	if err != nil {
 		return err
 	}
-	err = s.runContainer(ctx, project, service, name, &container)
+	number, err := strconv.Atoi(container.Labels["com.docker.compose.container-number"])
+	if err != nil {
+		return err
+	}
+	err = s.runContainer(ctx, project, service, name, number, &container)
 	if err != nil {
 		return err
 	}
@@ -143,8 +206,8 @@ func (s *local) restartContainer(ctx context.Context, service types.ServiceConfi
 	return nil
 }
 
-func (s *local) runContainer(ctx context.Context, project *types.Project, service types.ServiceConfig, name string, container *moby.Container) error {
-	containerConfig, hostConfig, networkingConfig, err := getContainerCreateOptions(project, service, container)
+func (s *local) runContainer(ctx context.Context, project *types.Project, service types.ServiceConfig, name string, number int, container *moby.Container) error {
+	containerConfig, hostConfig, networkingConfig, err := getContainerCreateOptions(project, service, number, container)
 	if err != nil {
 		return err
 	}