reaper_test.go

package testcontainers

import (
	"context"
	"errors"
	"fmt"
	"net"
	"os"
	"strconv"
	"sync"
	"syscall"
	"testing"
	"time"

	"github.com/cenkalti/backoff/v4"
	"github.com/containerd/errdefs"
	"github.com/moby/moby/api/types/container"
	"github.com/moby/moby/api/types/network"
	"github.com/stretchr/testify/require"

	"github.com/testcontainers/testcontainers-go/internal/config"
	"github.com/testcontainers/testcontainers-go/internal/core"
)

// testSessionID the tests need to create a reaper in a different session, so that it does not interfere with other tests
const testSessionID = "this-is-a-different-session-id"

type mockReaperProvider struct {
	req              ContainerRequest
	hostConfig       *container.HostConfig
	endpointSettings map[string]*network.EndpointSettings
	config           TestcontainersConfig
}

func newMockReaperProvider(cfg config.Config) *mockReaperProvider {
	m := &mockReaperProvider{
		config: TestcontainersConfig{
			Config: cfg,
		},
	}

	return m
}

var errExpected = errors.New("expected")

func (m *mockReaperProvider) RunContainer(_ context.Context, req ContainerRequest) (Container, error) {
	m.req = req

	m.hostConfig = &container.HostConfig{}
	m.endpointSettings = map[string]*network.EndpointSettings{}

	if req.HostConfigModifier == nil {
		req.HostConfigModifier = defaultHostConfigModifier(req)
	}
	req.HostConfigModifier(m.hostConfig)

	if req.EndpointSettingsModifier != nil {
		req.EndpointSettingsModifier(m.endpointSettings)
	}

	// we're only interested in the request, so instead of mocking the Docker client
	// we'll error out here
	return nil, errExpected
}

func (m *mockReaperProvider) Config() TestcontainersConfig {
	return m.config
}

// expectedReaperRequest creates the expected reaper container request with the given customizations.
func expectedReaperRequest(customize ...func(*ContainerRequest)) ContainerRequest {
	req := ContainerRequest{
		Image:        config.ReaperDefaultImage,
		ExposedPorts: []string{"8080/tcp"},
		Labels:       core.DefaultLabels(testSessionID),
		HostConfigModifier: func(hostConfig *container.HostConfig) {
			hostConfig.Binds = []string{core.MustExtractDockerSocket(context.Background()) + ":/var/run/docker.sock"}
			hostConfig.Privileged = true
		},
		WaitingFor: defaultRyukWaitStrategy("8080/tcp"),
		Env: map[string]string{
			"RYUK_CONNECTION_TIMEOUT":   "1m0s",
			"RYUK_RECONNECTION_TIMEOUT": "10s",
		},
	}

	req.Labels[core.LabelReaper] = "true"
	req.Labels[core.LabelRyuk] = "true"
	delete(req.Labels, core.LabelReap)

	for _, customize := range customize {
		customize(&req)
	}

	return req
}

// reaperDisable disables / enables the reaper for the duration of the test.
func reaperDisable(t *testing.T, disabled bool) {
	t.Helper()

	config.Reset()
	t.Setenv("TESTCONTAINERS_RYUK_DISABLED", strconv.FormatBool(disabled))
	t.Cleanup(config.Reset)
}

func testContainerStart(t *testing.T) {
	t.Helper()
	ctx := context.Background()

	ctr, err := Run(ctx, nginxAlpineImage, WithExposedPorts(nginxDefaultPort))
	CleanupContainer(t, ctr)
	require.NoError(t, err)
}

// testReaperRunning validates that a reaper is running.
func testReaperRunning(t *testing.T) {
	t.Helper()

	ctx := context.Background()
	sessionID := config.Read().SessionID
	reaperContainer, err := spawner.lookupContainer(ctx, sessionID)
	require.NoError(t, err)
	require.NotNil(t, reaperContainer)
}

func TestContainer(t *testing.T) {
	reaperDisable(t, false)

	t.Run("start/reaper-enabled", func(t *testing.T) {
		testContainerStart(t)
		testReaperRunning(t)
	})

	t.Run("stop/reaper-enabled", func(t *testing.T) {
		testContainerStop(t)
		testReaperRunning(t)
	})

	t.Run("terminate/reaper-enabled", func(t *testing.T) {
		testContainerTerminate(t)
		testReaperRunning(t)
	})

	reaperDisable(t, true)

	t.Run("start/reaper-disabled", func(t *testing.T) {
		testContainerStart(t)
	})

	t.Run("stop/reaper-disabled", func(t *testing.T) {
		testContainerStop(t)
	})

	t.Run("terminate/reaper-disabled", func(t *testing.T) {
		testContainerTerminate(t)
	})
}

// testContainerStop tests stopping a container.
func testContainerStop(t *testing.T) {
	t.Helper()

	ctx := context.Background()

	nginxA, err := Run(ctx, nginxAlpineImage, WithExposedPorts(nginxDefaultPort))
	CleanupContainer(t, nginxA)
	require.NoError(t, err)

	state, err := nginxA.State(ctx)
	require.NoError(t, err)
	require.True(t, state.Running)

	stopTimeout := 10 * time.Second
	err = nginxA.Stop(ctx, &stopTimeout)
	require.NoError(t, err)

	state, err = nginxA.State(ctx)
	require.NoError(t, err)
	require.False(t, state.Running)
	require.Equal(t, container.StateExited, state.Status)
}

// testContainerTerminate tests terminating a container.
func testContainerTerminate(t *testing.T) {
	t.Helper()
	ctx := context.Background()

	nginxA, err := Run(ctx, nginxAlpineImage, WithExposedPorts(nginxDefaultPort))
	CleanupContainer(t, nginxA)
	require.NoError(t, err)

	state, err := nginxA.State(ctx)
	require.NoError(t, err)
	require.True(t, state.Running)

	err = nginxA.Terminate(ctx)
	require.NoError(t, err)

	_, err = nginxA.State(ctx)
	require.Error(t, err)
}

func Test_NewReaper(t *testing.T) {
	reaperDisable(t, false)

	ctx := context.Background()

	t.Run("non-privileged", func(t *testing.T) {
		testNewReaper(ctx, t,
			config.Config{
				RyukConnectionTimeout:   time.Minute,
				RyukReconnectionTimeout: 10 * time.Second,
			},
			expectedReaperRequest(),
			false,
		)
	})

	t.Run("privileged", func(t *testing.T) {
		testNewReaper(ctx, t,
			config.Config{
				RyukPrivileged:          true,
				RyukConnectionTimeout:   time.Minute,
				RyukReconnectionTimeout: 10 * time.Second,
			},
			expectedReaperRequest(),
			true,
		)
	})

	t.Run("custom-timeouts", func(t *testing.T) {
		testNewReaper(ctx, t,
			config.Config{
				RyukPrivileged:          true,
				RyukConnectionTimeout:   2 * time.Minute,
				RyukReconnectionTimeout: 20 * time.Second,
			},
			expectedReaperRequest(func(req *ContainerRequest) {
				req.Env = map[string]string{
					"RYUK_CONNECTION_TIMEOUT":   "2m0s",
					"RYUK_RECONNECTION_TIMEOUT": "20s",
				}
			}),
			true,
		)
	})

	t.Run("verbose", func(t *testing.T) {
		testNewReaper(ctx, t,
			config.Config{
				RyukPrivileged: true,
				RyukVerbose:    true,
			},
			expectedReaperRequest(func(req *ContainerRequest) {
				req.Env = map[string]string{
					"RYUK_VERBOSE": "true",
				}
			}),
			true,
		)
	})

	t.Run("docker-host", func(t *testing.T) {
		testNewReaper(context.WithValue(ctx, core.DockerHostContextKey, core.DockerSocketPathWithSchema), t,
			config.Config{
				RyukConnectionTimeout:   time.Minute,
				RyukReconnectionTimeout: 10 * time.Second,
			},
			expectedReaperRequest(func(req *ContainerRequest) {
				req.HostConfigModifier = func(hostConfig *container.HostConfig) {
					hostConfig.Binds = []string{core.MustExtractDockerSocket(ctx) + ":/var/run/docker.sock"}
				}
			}),
			false,
		)
	})

	t.Run("hub-prefix", func(t *testing.T) {
		testNewReaper(context.WithValue(ctx, core.DockerHostContextKey, core.DockerSocketPathWithSchema), t,
			config.Config{
				HubImageNamePrefix:      "registry.mycompany.com/mirror",
				RyukPrivileged:          true,
				RyukConnectionTimeout:   time.Minute,
				RyukReconnectionTimeout: 10 * time.Second,
			},
			expectedReaperRequest(func(req *ContainerRequest) {
				req.Image = config.ReaperDefaultImage
				req.HostConfigModifier = func(hc *container.HostConfig) {
					hc.Privileged = true
				}
			}),
			true,
		)
	})

	t.Run("hub-prefix-env", func(t *testing.T) {
		config.Reset()
		t.Cleanup(config.Reset)

		t.Setenv("TESTCONTAINERS_HUB_IMAGE_NAME_PREFIX", "registry.mycompany.com/mirror")
		testNewReaper(context.WithValue(ctx, core.DockerHostContextKey, core.DockerSocketPathWithSchema), t,
			config.Config{
				RyukPrivileged:          true,
				RyukConnectionTimeout:   time.Minute,
				RyukReconnectionTimeout: 10 * time.Second,
			},
			expectedReaperRequest(func(req *ContainerRequest) {
				req.Image = config.ReaperDefaultImage
				req.HostConfigModifier = func(hc *container.HostConfig) {
					hc.Privileged = true
				}
			}),
			true,
		)
	})
}

func testNewReaper(ctx context.Context, t *testing.T, cfg config.Config, expected ContainerRequest, privileged bool) {
	t.Helper()

	if prefix := os.Getenv("TESTCONTAINERS_HUB_IMAGE_NAME_PREFIX"); prefix != "" {
		cfg.HubImageNamePrefix = prefix
	}

	provider := newMockReaperProvider(cfg)

	// We need a new reaperSpawner for each test case to avoid reusing
	// an existing reaper instance.
	spawner := &reaperSpawner{}
	reaper, err := spawner.reaper(ctx, testSessionID, provider)
	cleanupReaper(t, reaper, spawner)
	// We should have errored out see mockReaperProvider.RunContainer.
	require.ErrorIs(t, err, errExpected)

	require.Equal(t, expected.Image, provider.req.Image, "expected image doesn't match the submitted request")
	require.Equal(t, expected.ExposedPorts, provider.req.ExposedPorts, "expected exposed ports don't match the submitted request")
	require.Equal(t, expected.Labels, provider.req.Labels, "expected labels don't match the submitted request")
	require.Equal(t, expected.Mounts, provider.req.Mounts, "expected mounts don't match the submitted request")
	require.Equal(t, expected.WaitingFor, provider.req.WaitingFor, "expected waitingFor don't match the submitted request")
	require.Equal(t, expected.Env, provider.req.Env, "expected env doesn't match the submitted request")

	// checks for reaper's preCreationCallback fields
	require.Equal(t, container.NetworkMode(Bridge), provider.hostConfig.NetworkMode, "expected networkMode doesn't match the submitted request")
	require.True(t, provider.hostConfig.AutoRemove, "expected networkMode doesn't match the submitted request")
	require.Equal(t, privileged, provider.hostConfig.Privileged, "expected privileged doesn't match the submitted request")
}

func Test_ReaperReusedIfHealthy(t *testing.T) {
	reaperDisable(t, false)

	SkipIfProviderIsNotHealthy(t)

	ctx := context.Background()
	// As other integration tests run with the (shared) Reaper as well, re-use the instance to not interrupt other tests
	if spawner.instance != nil {
		t.Cleanup(func() {
			require.NoError(t, spawner.cleanup())
		})
	}

	provider, err := ProviderDocker.GetProvider()
	require.NoError(t, err)

	reaper, err := spawner.reaper(context.WithValue(ctx, core.DockerHostContextKey, provider.(*DockerProvider).host), testSessionID, provider)
	cleanupReaper(t, reaper, spawner)
	require.NoError(t, err, "creating the Reaper should not error")

	reaperReused, err := spawner.reaper(context.WithValue(ctx, core.DockerHostContextKey, provider.(*DockerProvider).host), testSessionID, provider)
	cleanupReaper(t, reaper, spawner)
	require.NoError(t, err, "reusing the Reaper should not error")

	// Ensure the internal state of both reaper instances is the same
	require.Equal(t, reaper.SessionID, reaperReused.SessionID, "expecting the same SessionID")
	require.Equal(t, reaper.Endpoint, reaperReused.Endpoint, "expecting the same reaper endpoint")
	require.Equal(t, reaper.Provider, reaperReused.Provider, "expecting the same container provider")
	require.Equal(t, reaper.container.GetContainerID(), reaperReused.container.GetContainerID(), "expecting the same container ID")
	require.Equal(t, reaper.container.SessionID(), reaperReused.container.SessionID(), "expecting the same session ID")

	termSignal, err := reaper.Connect()
	cleanupTermSignal(t, termSignal)
	require.NoError(t, err, "connecting to Reaper should be successful")
}

func Test_RecreateReaperIfTerminated(t *testing.T) {
	reaperDisable(t, false)

	SkipIfProviderIsNotHealthy(t)

	provider, err := ProviderDocker.GetProvider()
	require.NoError(t, err)

	ctx := context.Background()
	reaper, err := spawner.reaper(context.WithValue(ctx, core.DockerHostContextKey, provider.(*DockerProvider).host), testSessionID, provider)
	cleanupReaper(t, reaper, spawner)
	require.NoError(t, err, "creating the Reaper should not error")

	termSignal, err := reaper.Connect()
	if termSignal != nil {
		termSignal <- true
	}
	require.NoError(t, err)

	// Wait for up to ryuk's default reconnect timeout + 1s to allow for a graceful shutdown/cleanup of the container.
	timeout := time.NewTimer(time.Second * 11)
	t.Cleanup(func() {
		timeout.Stop()
	})
	for {
		state, err := reaper.container.State(ctx)
		if err != nil {
			if errdefs.IsNotFound(err) {
				break
			}
			require.NoError(t, err)
		}

		if !state.Running {
			break
		}

		select {
		case <-timeout.C:
			t.Fatal("reaper container should have been terminated")
		default:
		}

		time.Sleep(time.Millisecond * 100)
	}

	recreatedReaper, err := spawner.reaper(context.WithValue(ctx, core.DockerHostContextKey, provider.(*DockerProvider).host), testSessionID, provider)
	cleanupReaper(t, recreatedReaper, spawner)
	require.NoError(t, err, "creating the Reaper should not error")
	require.NotEqual(t, reaper.container.GetContainerID(), recreatedReaper.container.GetContainerID(), "expected different container ID")

	recreatedTermSignal, err := recreatedReaper.Connect()
	cleanupTermSignal(t, recreatedTermSignal)
	require.NoError(t, err, "connecting to Reaper should be successful")
}

func TestReaper_reuseItFromOtherTestProgramUsingDocker(t *testing.T) {
	reaperDisable(t, false)

	// Explicitly set the reaper instance to nil to simulate another test
	// program in the same session accessing the same reaper.
	spawner.instance = nil

	SkipIfProviderIsNotHealthy(t)

	ctx := context.Background()
	// As other integration tests run with the (shared) Reaper as well,
	// re-use the instance to not interrupt other tests.
	if spawner.instance != nil {
		t.Cleanup(func() {
			require.NoError(t, spawner.cleanup())
		})
	}

	provider, err := ProviderDocker.GetProvider()
	require.NoError(t, err)

	reaper, err := spawner.reaper(context.WithValue(ctx, core.DockerHostContextKey, provider.(*DockerProvider).host), testSessionID, provider)
	cleanupReaper(t, reaper, spawner)
	require.NoError(t, err, "creating the Reaper should not error")

	// Explicitly reset the reaper instance to nil to simulate another test
	// program in the same session accessing the same reaper.
	spawner.instance = nil

	reaperReused, err := spawner.reaper(context.WithValue(ctx, core.DockerHostContextKey, provider.(*DockerProvider).host), testSessionID, provider)
	cleanupReaper(t, reaper, spawner)
	require.NoError(t, err, "reusing the Reaper should not error")

	// Ensure that the internal state of both reaper instances is the same.
	require.Equal(t, reaper.SessionID, reaperReused.SessionID, "expecting the same SessionID")
	require.Equal(t, reaper.Endpoint, reaperReused.Endpoint, "expecting the same reaper endpoint")
	require.Equal(t, reaper.Provider, reaperReused.Provider, "expecting the same container provider")
	require.Equal(t, reaper.container.GetContainerID(), reaperReused.container.GetContainerID(), "expecting the same container ID")
	require.Equal(t, reaper.container.SessionID(), reaperReused.container.SessionID(), "expecting the same session ID")

	termSignal, err := reaper.Connect()
	cleanupTermSignal(t, termSignal)
	require.NoError(t, err, "connecting to Reaper should be successful")
}

// TestReaper_ReuseRunning tests whether reusing the reaper if using
// testcontainers from concurrently multiple packages works as expected. In this
// case, global locks are without any effect as Go tests different packages
// isolated. Therefore, this test does not use the same logic with locks on
// purpose. We expect reaper creation to still succeed in case a reaper is
// already running for the same session id by returning its container instance
// instead.
func TestReaper_ReuseRunning(t *testing.T) {
	reaperDisable(t, false)

	const concurrency = 64

	timeout, cancel := context.WithTimeout(context.Background(), time.Minute)
	defer cancel()

	sessionID := SessionID()

	dockerProvider, err := NewDockerProvider()
	require.NoError(t, err, "new docker provider should not fail")

	obtainedReaperContainerIDs := make([]string, concurrency)
	var wg sync.WaitGroup
	for i := range concurrency {
		wg.Add(1)
		go func(i int) {
			defer wg.Done()
			spawner := &reaperSpawner{}
			reaper, err := spawner.reaper(timeout, sessionID, dockerProvider)
			cleanupReaper(t, reaper, spawner)
			require.NoError(t, err)

			reaperConnect(t, reaper)

			obtainedReaperContainerIDs[i] = reaper.container.GetContainerID()
		}(i)
	}
	wg.Wait()

	// Assure that all calls returned the same container.
	firstContainerID := obtainedReaperContainerIDs[0]
	for i, containerID := range obtainedReaperContainerIDs {
		require.Equal(t, firstContainerID, containerID, "call %d should have returned same container id", i)
	}
}

// reaperConnect copies the logic from Reaper.connect() but with better error handling.
// Reaper.connect() neither returns the error from the handshake, nor the connection,
// making the testing of the handshake flow impossible.
func reaperConnect(t *testing.T, reaper *Reaper) {
	t.Helper()

	ctx, cancel := context.WithTimeout(context.Background(), time.Second*5)
	defer cancel()
	var d net.Dialer
	conn, err := d.DialContext(ctx, "tcp", reaper.Endpoint)
	require.NoError(t, err, "dial reaper %s: %v", reaper.Endpoint, err)
	defer conn.Close()
	err = reaper.handshake(conn)
	require.NoError(t, err, "Reaper handshake should be successful")
}

func TestSpawnerBackoff(t *testing.T) {
	b := spawner.backoff()
	for range 100 {
		require.LessOrEqual(t, b.NextBackOff(), time.Millisecond*250, "backoff should not exceed max interval")
	}
}

type timeoutErr struct{}

func (timeoutErr) Error() string {
	return "timeout"
}

func (timeoutErr) Timeout() bool {
	return true
}

func TestSpawnerRetryError(t *testing.T) {
	t.Run("nil error", func(t *testing.T) {
		err := spawner.retryError(nil)
		require.NoError(t, err, "should return nil")
	})

	tests := []struct {
		name      string
		err       error
		permanent bool
	}{
		{
			name:      "docker conflict error",
			err:       errors.New("Conflict. The container name \"foo\" is already in use by container \"01234\"."),
			permanent: false,
		},
		{
			name:      "podman conflict error",
			err:       errors.New("creating container storage: the container name \"foo\" is already in use by 01234."),
			permanent: false,
		},
		{
			name:      "errdefs.ErrNotFound",
			err:       fmt.Errorf("foo: %w", errdefs.ErrNotFound),
			permanent: false,
		},
		{
			name:      "errdefs.Conflict",
			err:       fmt.Errorf("foo: %w", errdefs.ErrConflict),
			permanent: true,
		},
		{
			name:      "syscall.ECONNREFUSED",
			err:       fmt.Errorf("foo: %w", syscall.ECONNREFUSED),
			permanent: false,
		},
		{
			name:      "syscall.ECONNRESET",
			err:       fmt.Errorf("foo: %w", syscall.ECONNRESET),
			permanent: false,
		},
		{
			name:      "syscall.ECONNABORTED",
			err:       fmt.Errorf("foo: %w", syscall.ECONNABORTED),
			permanent: false,
		},
		{
			name:      "syscall.ETIMEDOUT",
			err:       fmt.Errorf("foo: %w", syscall.ETIMEDOUT),
			permanent: false,
		},
		{
			name:      "os.ErrDeadlineExceeded",
			err:       fmt.Errorf("foo: %w", os.ErrDeadlineExceeded),
			permanent: false,
		},
		{
			name:      "context.DeadlineExceeded",
			err:       fmt.Errorf("foo: %w", context.DeadlineExceeded),
			permanent: false,
		},
		{
			name:      "timeout error",
			err:       fmt.Errorf("foo: %w", timeoutErr{}),
			permanent: false,
		},
		{
			name:      "context.Canceled",
			err:       fmt.Errorf("foo: %w", context.Canceled),
			permanent: false,
		},
		{
			name:      "random error",
			err:       errors.New("some random error"),
			permanent: true,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			gotErr := spawner.retryError(tt.err)
			require.Error(t, gotErr, "should not return nil")

			permanentError := &backoff.PermanentError{}
			isPermanent := errors.As(gotErr, &permanentError)
			if tt.permanent {
				require.True(t, isPermanent, "the error should be a PermanentError")
			} else {
				require.False(t, isPermanent, "the error should not be a PermanentError")
			}
		})
	}
}

// cleanupReaper schedules reaper for cleanup if it's not nil.
func cleanupReaper(t *testing.T, reaper *Reaper, spawner *reaperSpawner) {
	t.Helper()

	if reaper == nil {
		return
	}

	t.Cleanup(func() {
		reaper.close()
		require.NoError(t, spawner.cleanup())
	})
}

// cleanupTermSignal ensures that termSignal
func cleanupTermSignal(t *testing.T, termSignal chan bool) {
	t.Helper()

	t.Cleanup(func() {
		if termSignal != nil {
			termSignal <- true
		}
	})
}