Fix race condition when looking up reaper (ryuk) container (#2508)

* If the ryuk container has a health status, wait for a healthy container before returning.

* Use docker/types for health status, but also check for the zero value.

* Wait for the ryuk port to be available in case the container is still being started by newReaper in a separate process.

A race between newReaper and lookUpReaperContainer occurs:
- newReaper creates the container with a ContainerRequest.WaitingFor = wait.ForListeningPort(listeningPort)
- newReaper starts the container
- lookUpReaperContainer obtains the container and returns.
- newReaper invokes the readiness hook wait.ForListeningPort(listeningPort).

* Fix whitespace.

* chore: simplify

* chore: make lint

* chore: change emoji in log output when waiting

* chore: move inside reuseReaper

---------

Co-authored-by: Manuel de la Peña <mdelapenya@gmail.com>

docker.go

@@ -45,6 +45,7 @@ type DockerContainer struct {
 	sessionID         string
 	terminationSignal chan bool
 	keepBuiltImage    bool
+	healthStatus      string // container health status, will default to healthStatusNone if no healthcheck is present
 	// log consumer fields
 	consumers []log.Consumer
 	// TODO: Remove locking and wait group once the deprecated StartLogProducer and
@@ -92,6 +93,11 @@ func containerFromDockerResponse(ctx context.Context, response types.Container)
 		return nil, err
 	}
 
+	// the health status of the container, if any
+	if health := container.raw.State.Health; health != nil {
+		container.healthStatus = health.Status
+	}
+
 	return &container, nil
 }
 
@@ -720,7 +726,7 @@ func (c *DockerContainer) WaitUntilReady(ctx context.Context) error {
 
 	// if a Wait Strategy has been specified, wait before returning
 	c.Printf(
-		"🚧 Waiting for container id %s image: %s. Waiting for: %+v",
+		"⏳ Waiting for container id %s image: %s. Waiting for: %+v",
 		c.GetContainerID()[:12], c.GetImage(), c.WaitingFor,
 	)
 

generic_test.go

@@ -125,7 +125,7 @@ func TestNewReusableContainerInSubprocess(t *testing.T) {
 			output := createReuseContainerInSubprocess(t)
 
 			// check is reuse container with WaitingFor work correctly.
-			require.True(t, strings.Contains(output, "🚧 Waiting for container id"))
+			require.True(t, strings.Contains(output, "⏳ Waiting for container id"))
 			require.True(t, strings.Contains(output, "🔔 Container is ready"))
 		}()
 	}

reaper.go

@@ -8,6 +8,7 @@ import (
 	"time"
 
 	"github.com/cenkalti/backoff/v4"
+	"github.com/docker/docker/api/types"
 	"github.com/docker/docker/api/types/container"
 	"github.com/docker/docker/api/types/filters"
 	"github.com/docker/docker/errdefs"
@@ -274,6 +275,15 @@ func lookUpReaperContainer(ctx context.Context, sessionID string) (*DockerContai
 
 		reaperContainer = r
 
+		if r.healthStatus == types.Healthy || r.healthStatus == types.NoHealthcheck {
+			return nil
+		}
+
+		// if a health status is present on the container, and the container is healthy, error
+		if r.healthStatus != "" {
+			return fmt.Errorf("container %s is not healthy, wanted status=%s, got status=%s", resp[0].ID[:8], types.Healthy, r.healthStatus)
+		}
+
 		return nil
 	}, backoff.WithContext(exp, ctx))
 	if err != nil {
@@ -299,6 +309,26 @@ func reuseReaperContainer(ctx context.Context, sessionID string, reaperContainer
 		return nil, err
 	}
 
+	reaperContainer.Printf("⏳ Waiting for Reaper port to be ready")
+
+	var containerJson *types.ContainerJSON
+
+	if containerJson, err = reaperContainer.Inspect(ctx); err != nil {
+		return nil, fmt.Errorf("failed to inspect reaper container %s: %w", reaperContainer.ID[:8], err)
+	}
+
+	if containerJson != nil && containerJson.NetworkSettings != nil {
+		for port := range containerJson.NetworkSettings.Ports {
+			err := wait.ForListeningPort(port).
+				WithPollInterval(100*time.Millisecond).
+				WaitUntilReady(ctx, reaperContainer)
+			if err != nil {
+				return nil, fmt.Errorf("failed waiting for reaper container %s port %s/%s to be ready: %w",
+					reaperContainer.ID[:8], port.Proto(), port.Port(), err)
+			}
+		}
+	}
+
 	return &Reaper{
 		SessionID: sessionID,
 		Endpoint:  endpoint,