re-implement compose logic

- build observed state
- compute reconciliation plan by comparing observed vs desired state
- execute plan

this allow to decorelate reconciliation (aka "convergence") logic from docker API,
and write simpler and efficient tests to cover various scenarios

Signed-off-by: Nicolas De Loof <nicolas.deloof@gmail.com>

Author: ndeloof

pkg/compose/convergence.go

@@ -87,82 +87,116 @@ func (s *composeService) create(ctx context.Context, project *types.Project, opt
 
 	prepareNetworks(project)
 
-	networks, err := s.ensureNetworks(ctx, project)
+	// Temporary implementation of use_api_socket until we get actual support inside docker engine
+	project, err = s.useAPISocket(project)
 	if err != nil {
 		return err
 	}
 
-	volumes, err := s.ensureProjectVolumes(ctx, project)
+	// Phase 1: Inspect current state
+	observed, err := s.InspectState(ctx, project)
 	if err != nil {
 		return err
 	}
 
-	var observedState Containers
-	observedState, err = s.getContainers(ctx, project.Name, oneOffInclude, true)
-	if err != nil {
-		return err
-	}
-	orphans := observedState.filter(isOrphaned(project))
-	if len(orphans) > 0 && !options.IgnoreOrphans {
-		if options.RemoveOrphans {
-			err := s.removeContainers(ctx, orphans, nil, nil, false)
-			if err != nil {
-				return err
-			}
-		} else {
+	// Handle orphan containers
+	if len(observed.Orphans) > 0 && !options.IgnoreOrphans {
+		if !options.RemoveOrphans {
 			logrus.Warnf("Found orphan containers (%s) for this project. If "+
 				"you removed or renamed this service in your compose "+
 				"file, you can run this command with the "+
-				"--remove-orphans flag to clean it up.", orphans.names())
+				"--remove-orphans flag to clean it up.", observed.Orphans.names())
 		}
 	}
 
-	// Temporary implementation of use_api_socket until we get actual support inside docker engine
-	project, err = s.useAPISocket(project)
-	if err != nil {
+	// Validate external networks exist before reconciling
+	if err := s.validateExternalNetworks(ctx, project, options.Services); err != nil {
 		return err
 	}
 
-	return newConvergence(options.Services, observedState, networks, volumes, s).apply(ctx, project, options)
-}
+	// Phase 2: Reconcile desired vs observed state (pure function)
+	plan, err := Reconcile(project, observed, ReconcileOptions{
+		Recreate:             options.Recreate,
+		RecreateDependencies: options.RecreateDependencies,
+		Services:             options.Services,
+		Inherit:              options.Inherit,
+		Timeout:              options.Timeout,
+		RemoveOrphans:        options.RemoveOrphans,
+	})
+	if err != nil {
+		return err
+	}
 
-func prepareNetworks(project *types.Project) {
-	for k, nw := range project.Networks {
-		nw.CustomLabels = nw.CustomLabels.
-			Add(api.NetworkLabel, k).
-			Add(api.ProjectLabel, project.Name).
-			Add(api.VersionLabel, api.ComposeVersion)
-		project.Networks[k] = nw
+	if plan.IsEmpty() {
+		return nil
 	}
+
+	s.emitUntouchedContainerEvents(project, observed, plan)
+
+	// Phase 3: Execute the plan
+	return s.ExecutePlan(ctx, project, plan)
 }
 
-func (s *composeService) ensureNetworks(ctx context.Context, project *types.Project) (map[string]string, error) {
-	networks := map[string]string{}
-	for name, nw := range project.Networks {
-		id, err := s.ensureNetwork(ctx, project, name, &nw)
-		if err != nil {
-			return nil, err
+// emitUntouchedContainerEvents emits progress events for containers that are
+// already up-to-date and running, so that callers (e.g. scale) can see them.
+func (s *composeService) emitUntouchedContainerEvents(project *types.Project, observed *ObservedState, plan *ReconciliationPlan) {
+	for _, service := range project.Services {
+		for _, ctr := range observed.Containers[service.Name] {
+			ctrName := getCanonicalContainerName(ctr)
+			if _, touched := plan.Operations["recreate-container:"+ctrName]; touched {
+				continue
+			}
+			if _, touched := plan.Operations["stop-container:"+ctrName]; touched {
+				continue
+			}
+			if _, touched := plan.Operations["create-container:"+ctrName]; touched {
+				continue
+			}
+			if ctr.State == container.StateRunning {
+				s.events.On(runningEvent(getContainerProgressName(ctr)))
+			}
 		}
-		networks[name] = id
-		project.Networks[name] = nw
 	}
-	return networks, nil
 }
 
-func (s *composeService) ensureProjectVolumes(ctx context.Context, project *types.Project) (map[string]string, error) {
-	ids := map[string]string{}
-	for k, volume := range project.Volumes {
-		volume.CustomLabels = volume.CustomLabels.Add(api.VolumeLabel, k)
-		volume.CustomLabels = volume.CustomLabels.Add(api.ProjectLabel, project.Name)
-		volume.CustomLabels = volume.CustomLabels.Add(api.VersionLabel, api.ComposeVersion)
-		id, err := s.ensureVolume(ctx, k, volume, project)
+// validateExternalNetworks checks that external networks exist for services
+// that are part of the current operation. Returns an error if a required
+// external network is not found.
+func (s *composeService) validateExternalNetworks(ctx context.Context, project *types.Project, services []string) error {
+	for key, net := range project.Networks {
+		if !net.External {
+			continue
+		}
+		// Check if any targeted service uses this network
+		usedByTargetedService := false
+		for _, service := range project.Services {
+			if len(services) > 0 && !slices.Contains(services, service.Name) {
+				continue
+			}
+			if _, ok := service.Networks[key]; ok {
+				usedByTargetedService = true
+				break
+			}
+		}
+		if !usedByTargetedService {
+			continue
+		}
+		_, err := s.resolveExternalNetwork(ctx, &net)
 		if err != nil {
-			return nil, err
+			return err
 		}
-		ids[k] = id
 	}
+	return nil
+}
 
-	return ids, nil
+func prepareNetworks(project *types.Project) {
+	for k, nw := range project.Networks {
+		nw.CustomLabels = nw.CustomLabels.
+			Add(api.NetworkLabel, k).
+			Add(api.ProjectLabel, project.Name).
+			Add(api.VersionLabel, api.ComposeVersion)
+		project.Networks[k] = nw
+	}
 }
 
 //nolint:gocyclo

pkg/compose/create.go

@@ -0,0 +1,143 @@
+/*
+   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 (
+	"context"
+
+	"github.com/compose-spec/compose-go/v2/types"
+	"github.com/moby/moby/client"
+	"golang.org/x/sync/errgroup"
+
+	"github.com/docker/compose/v5/pkg/api"
+)
+
+// ObservedState captures the current state of a Compose project as seen by the Docker engine.
+type ObservedState struct {
+	ProjectName string
+	Containers  map[string]Containers
+	Networks    map[string]ObservedNetwork
+	Volumes     map[string]ObservedVolume
+	Orphans     Containers
+}
+
+// ObservedNetwork represents a Docker network associated with a Compose project.
+type ObservedNetwork struct {
+	ID         string
+	Name       string
+	Driver     string
+	Labels     map[string]string
+	ConfigHash string
+}
+
+// ObservedVolume represents a Docker volume associated with a Compose project.
+type ObservedVolume struct {
+	Name       string
+	Driver     string
+	Labels     map[string]string
+	ConfigHash string
+}
+
+// allContainers returns all containers from the observed state (across all services and orphans).
+func (s *ObservedState) allContainers() Containers {
+	var result Containers
+	for _, ctrs := range s.Containers {
+		result = append(result, ctrs...)
+	}
+	result = append(result, s.Orphans...)
+	return result
+}
+
+// InspectState queries the Docker engine to build an ObservedState for the given project.
+func (s *composeService) InspectState(ctx context.Context, project *types.Project) (*ObservedState, error) {
+	var (
+		allContainers Containers
+		networks      client.NetworkListResult
+		volumes       client.VolumeListResult
+	)
+
+	eg, ctx := errgroup.WithContext(ctx)
+
+	eg.Go(func() error {
+		var err error
+		allContainers, err = s.getContainers(ctx, project.Name, oneOffInclude, true)
+		return err
+	})
+
+	eg.Go(func() error {
+		var err error
+		networks, err = s.apiClient().NetworkList(ctx, client.NetworkListOptions{
+			Filters: projectFilter(project.Name),
+		})
+		return err
+	})
+
+	eg.Go(func() error {
+		var err error
+		volumes, err = s.apiClient().VolumeList(ctx, client.VolumeListOptions{
+			Filters: projectFilter(project.Name),
+		})
+		return err
+	})
+
+	if err := eg.Wait(); err != nil {
+		return nil, err
+	}
+
+	// Partition containers by service
+	containersByService := map[string]Containers{}
+	for _, c := range allContainers {
+		service := c.Labels[api.ServiceLabel]
+		containersByService[service] = append(containersByService[service], c)
+	}
+
+	// Identify orphan containers
+	orphans := allContainers.filter(isOrphaned(project))
+
+	// Map networks by their Compose network name
+	observedNetworks := map[string]ObservedNetwork{}
+	for _, n := range networks.Items {
+		name := n.Labels[api.NetworkLabel]
+		observedNetworks[name] = ObservedNetwork{
+			ID:         n.ID,
+			Name:       n.Name,
+			Driver:     n.Driver,
+			Labels:     n.Labels,
+			ConfigHash: n.Labels[api.ConfigHashLabel],
+		}
+	}
+
+	// Map volumes by their Compose volume name
+	observedVolumes := map[string]ObservedVolume{}
+	for _, v := range volumes.Items {
+		name := v.Labels[api.VolumeLabel]
+		observedVolumes[name] = ObservedVolume{
+			Name:       v.Name,
+			Driver:     v.Driver,
+			Labels:     v.Labels,
+			ConfigHash: v.Labels[api.ConfigHashLabel],
+		}
+	}
+
+	return &ObservedState{
+		ProjectName: project.Name,
+		Containers:  containersByService,
+		Networks:    observedNetworks,
+		Volumes:     observedVolumes,
+		Orphans:     orphans,
+	}, nil
+}