feat: support Ryuk for the compose module (testcontainers#2485)

* feat: add testcontainers labels to compose containers

* feat: support reaper for compose

* chore: increase ryuk reconnection timeout on CI

* chore: cache containers on UP

* chore: more tuning for compose

* chore: more consistent assertion

* chore: the compose stack asks for the reaper, but each container then connects to it

* chore: use different error groups

the first time wait is called, the context is cancelled

* chore: the lookup method include cache checks

* chore: update tests to make them deterministic

* chore: rename local compose testss

* chore: support returning the dynamic port in the helper function

* chore: try with default reconnection timeout

* feat: support removing networks from compose

* chore: support naming test services with local and api

It will allow the tests to be more deterministic, as there could be service containers started from the local test suite with the same name as in the API test suite.

* Revert "chore: try with default reconnection timeout"

This reverts commit 336760c.

* fix: typo

.github/workflows/ci-test-go.yml

@@ -50,6 +50,8 @@ jobs:
     continue-on-error: ${{ !inputs.fail-fast }}
     env:
       TESTCONTAINERS_RYUK_DISABLED: "${{ inputs.ryuk-disabled }}"
+      RYUK_CONNECTION_TIMEOUT: "${{ inputs.project-directory == 'modules/compose' && '5m' || '60s' }}"
+      RYUK_RECONNECTION_TIMEOUT: "${{ inputs.project-directory == 'modules/compose' && '30s' || '10s' }}"
     steps:
       - name: Setup rootless Docker
         if: ${{ inputs.rootless-docker }}

docker.go

@@ -98,6 +98,11 @@ func (c *DockerContainer) SetProvider(provider *DockerProvider) {
 	c.provider = provider
 }
 
+// SetTerminationSignal sets the termination signal for the container
+func (c *DockerContainer) SetTerminationSignal(signal chan bool) {
+	c.terminationSignal = signal
+}
+
 func (c *DockerContainer) GetContainerID() string {
 	return c.ID
 }
@@ -846,6 +851,10 @@ func (n *DockerNetwork) Remove(ctx context.Context) error {
 	return n.provider.client.NetworkRemove(ctx, n.ID)
 }
 
+func (n *DockerNetwork) SetTerminationSignal(signal chan bool) {
+	n.terminationSignal = signal
+}
+
 // DockerProvider implements the ContainerProvider interface
 type DockerProvider struct {
 	*DockerProviderOptions

modules/compose/compose.go

@@ -3,6 +3,7 @@ package compose
 import (
 	"context"
 	"errors"
+	"fmt"
 	"path/filepath"
 	"runtime"
 	"strings"
@@ -121,6 +122,25 @@ func NewDockerComposeWith(opts ...ComposeStackOption) (*dockerCompose, error) {
 		return nil, err
 	}
 
+	reaperProvider, err := testcontainers.NewDockerProvider()
+	if err != nil {
+		return nil, fmt.Errorf("failed to create reaper provider for compose: %w", err)
+	}
+
+	tcConfig := reaperProvider.Config()
+
+	var composeReaper *testcontainers.Reaper
+	if !tcConfig.RyukDisabled {
+		// NewReaper is deprecated: we need to find a way to create the reaper for compose
+		// bypassing the deprecation.
+		r, err := testcontainers.NewReaper(context.Background(), testcontainers.SessionID(), reaperProvider, "")
+		if err != nil {
+			return nil, fmt.Errorf("failed to create reaper for compose: %w", err)
+		}
+
+		composeReaper = r
+	}
+
 	composeAPI := &dockerCompose{
 		name:           composeOptions.Identifier,
 		configs:        composeOptions.Paths,
@@ -129,6 +149,9 @@ func NewDockerComposeWith(opts ...ComposeStackOption) (*dockerCompose, error) {
 		dockerClient:   dockerCli.Client(),
 		waitStrategies: make(map[string]wait.Strategy),
 		containers:     make(map[string]*testcontainers.DockerContainer),
+		networks:       make(map[string]*testcontainers.DockerNetwork),
+		sessionID:      testcontainers.SessionID(),
+		reaper:         composeReaper,
 	}
 
 	return composeAPI, nil

modules/compose/compose_api.go

@@ -11,6 +11,7 @@ import (
 	"github.com/compose-spec/compose-go/v2/types"
 	"github.com/docker/cli/cli/command"
 	"github.com/docker/compose/v2/pkg/api"
+	dockertypes "github.com/docker/docker/api/types"
 	"github.com/docker/docker/api/types/container"
 	"github.com/docker/docker/api/types/filters"
 	"github.com/docker/docker/client"
@@ -134,6 +135,9 @@ type dockerCompose struct {
 	// used in ServiceContainer(...) function to avoid calls to the Docker API
 	containers map[string]*testcontainers.DockerContainer
 
+	// cache for networks in the compose stack
+	networks map[string]*testcontainers.DockerNetwork
+
 	// docker/compose API service instance used to control the compose stack
 	composeService api.Service
 
@@ -147,6 +151,12 @@ type dockerCompose struct {
 	// compiled compose project
 	// can be nil if the stack wasn't started yet
 	project *types.Project
+
+	// sessionID is used to identify the reaper session
+	sessionID string
+
+	// reaper is used to clean up containers after the stack is stopped
+	reaper *testcontainers.Reaper
 }
 
 func (d *dockerCompose) ServiceContainer(ctx context.Context, svcName string) (*testcontainers.DockerContainer, error) {
@@ -235,26 +245,89 @@ func (d *dockerCompose) Up(ctx context.Context, opts ...StackUpOption) error {
 		return err
 	}
 
+	err = d.lookupNetworks(ctx)
+	if err != nil {
+		return err
+	}
+
+	if d.reaper != nil {
+		for _, n := range d.networks {
+			termSignal, err := d.reaper.Connect()
+			if err != nil {
+				return fmt.Errorf("failed to connect to reaper: %w", err)
+			}
+			n.SetTerminationSignal(termSignal)
+
+			// Cleanup on error, otherwise set termSignal to nil before successful return.
+			defer func() {
+				if termSignal != nil {
+					termSignal <- true
+				}
+			}()
+		}
+	}
+
+	errGrpContainers, errGrpCtx := errgroup.WithContext(ctx)
+
+	for _, srv := range d.project.Services {
+		// we are going to connect each container to the reaper
+		srv := srv
+		errGrpContainers.Go(func() error {
+			dc, err := d.lookupContainer(errGrpCtx, srv.Name)
+			if err != nil {
+				return err
+			}
+
+			if d.reaper != nil {
+				termSignal, err := d.reaper.Connect()
+				if err != nil {
+					return fmt.Errorf("failed to connect to reaper: %w", err)
+				}
+				dc.SetTerminationSignal(termSignal)
+
+				// Cleanup on error, otherwise set termSignal to nil before successful return.
+				defer func() {
+					if termSignal != nil {
+						termSignal <- true
+					}
+				}()
+			}
+
+			d.containers[srv.Name] = dc
+
+			return nil
+		})
+	}
+
+	// wait here for the containers lookup to finish
+	if err := errGrpContainers.Wait(); err != nil {
+		return err
+	}
+
 	if len(d.waitStrategies) == 0 {
 		return nil
 	}
 
-	errGrp, errGrpCtx := errgroup.WithContext(ctx)
+	errGrpWait, errGrpCtx := errgroup.WithContext(ctx)
 
 	for svc, strategy := range d.waitStrategies { // pinning the variables
 		svc := svc
 		strategy := strategy
 
-		errGrp.Go(func() error {
+		errGrpWait.Go(func() error {
 			target, err := d.lookupContainer(errGrpCtx, svc)
 			if err != nil {
 				return err
 			}
+
+			// cache all the containers on compose.up
+			d.containers[svc] = target
+
 			return strategy.WaitUntilReady(errGrpCtx, target)
 		})
 	}
 
-	return errGrp.Wait()
+	return errGrpWait.Wait()
 }
 
 func (d *dockerCompose) WaitForService(s string, strategy wait.Strategy) ComposeStack {
@@ -327,6 +400,34 @@ func (d *dockerCompose) lookupContainer(ctx context.Context, svcName string) (*t
 	return container, nil
 }
 
+func (d *dockerCompose) lookupNetworks(ctx context.Context) error {
+	d.containersLock.Lock()
+	defer d.containersLock.Unlock()
+
+	listOptions := dockertypes.NetworkListOptions{
+		Filters: filters.NewArgs(
+			filters.Arg("label", fmt.Sprintf("%s=%s", api.ProjectLabel, d.name)),
+		),
+	}
+
+	networks, err := d.dockerClient.NetworkList(ctx, listOptions)
+	if err != nil {
+		return err
+	}
+
+	for _, n := range networks {
+		dn := &testcontainers.DockerNetwork{
+			ID:     n.ID,
+			Name:   n.Name,
+			Driver: n.Driver,
+		}
+
+		d.networks[n.ID] = dn
+	}
+
+	return nil
+}
+
 func (d *dockerCompose) compileProject(ctx context.Context) (*types.Project, error) {
 	const nameAndDefaultConfigPath = 2
 	projectOptions := make([]cli.ProjectOptionsFn, len(d.projectOptions), len(d.projectOptions)+nameAndDefaultConfigPath)
@@ -353,6 +454,11 @@ func (d *dockerCompose) compileProject(ctx context.Context) (*types.Project, err
 			api.ConfigFilesLabel: strings.Join(proj.ComposeFiles, ","),
 			api.OneoffLabel:      "False", // default, will be overridden by `run` command
 		}
+
+		for k, label := range testcontainers.GenericLabels() {
+			s.CustomLabels[k] = label
+		}
+
 		for i, envFile := range compiledOptions.EnvFiles {
 			// add a label for each env file, indexed by its position
 			s.CustomLabels[fmt.Sprintf("%s.%d", api.EnvironmentFileLabel, i)] = envFile
@@ -361,6 +467,20 @@ func (d *dockerCompose) compileProject(ctx context.Context) (*types.Project, err
 		proj.Services[i] = s
 	}
 
+	for key, n := range proj.Networks {
+		n.Labels = map[string]string{
+			api.ProjectLabel: proj.Name,
+			api.NetworkLabel: n.Name,
+			api.VersionLabel: api.ComposeVersion,
+		}
+
+		for k, label := range testcontainers.GenericLabels() {
+			n.Labels[k] = label
+		}
+
+		proj.Networks[key] = n
+	}
+
 	return proj, nil
 }