fix: fix log producer stopping block indefinitely

docker.go

@@ -68,10 +68,13 @@ type DockerContainer struct {
 	terminationSignal chan bool
 	consumers         []LogConsumer
 	raw               *types.ContainerJSON
-	stopProducer      chan bool
-	producerDone      chan bool
-	logger            Logging
-	lifecycleHooks    []ContainerLifecycleHooks
+	// stopProducer should be closed when the current log producer should be stopped.
+	// This allows the producer to exit in case of error while still avoiding
+	// StopLogProducer blocking indefinitely.
+	stopProducer   chan bool
+	producerDone   chan bool
+	logger         Logging
+	lifecycleHooks []ContainerLifecycleHooks
 }
 
 // SetLogger sets the logger for the container
@@ -718,7 +721,7 @@ func (c *DockerContainer) StartLogProducer(ctx context.Context) error {
 // and sending them to each added LogConsumer
 func (c *DockerContainer) StopLogProducer() error {
 	if c.stopProducer != nil {
-		c.stopProducer <- true
+		close(c.stopProducer)
 		// block until the producer is actually done in order to avoid strange races
 		<-c.producerDone
 		c.stopProducer = nil

logconsumer_test.go

@@ -289,6 +289,79 @@ func Test_StartStop(t *testing.T) {
 	assert.Nil(t, c.Terminate(ctx))
 }
 
+// Test_StopOnCrashedProducer tests similarly to Test_StartStop but crashes the
+// container during the test.
+func Test_StopOnCrashedProducer(t *testing.T) {
+	ctx := context.Background()
+	req := ContainerRequest{
+		FromDockerfile: FromDockerfile{
+			Context:    "./testdata/",
+			Dockerfile: "echoserver.Dockerfile",
+		},
+		ExposedPorts: []string{"8080/tcp"},
+		WaitingFor:   wait.ForLog("ready"),
+	}
+
+	gReq := GenericContainerRequest{
+		ContainerRequest: req,
+		Started:          true,
+	}
+
+	c, err := GenericContainer(ctx, gReq)
+	require.NoError(t, err)
+
+	ep, err := c.Endpoint(ctx, "http")
+	require.NoError(t, err)
+
+	g := TestLogConsumer{
+		Msgs:     []string{},
+		Done:     make(chan bool),
+		Accepted: make(chan string),
+	}
+	c.FollowOutput(&g)
+
+	require.NoError(t, c.StopLogProducer(), "nothing should happen even if the producer is not started")
+
+	require.NoError(t, c.StartLogProducer(ctx))
+	require.Equal(t, <-g.Accepted, "ready\n")
+
+	require.Error(t, c.StartLogProducer(ctx), "log producer is already started")
+
+	_, err = http.Get(ep + "/stdout?echo=mlem")
+	require.NoError(t, err)
+	require.Equal(t, <-g.Accepted, "echo mlem\n")
+
+	require.NoError(t, c.StopLogProducer())
+
+	require.NoError(t, c.StartLogProducer(ctx))
+	require.Equal(t, <-g.Accepted, "ready\n")
+	require.Equal(t, <-g.Accepted, "echo mlem\n")
+
+	_, err = http.Get(ep + "/stdout?echo=mlem2")
+	require.NoError(t, err)
+	require.Equal(t, <-g.Accepted, "echo mlem2\n")
+
+	// Here, we crash the container which should make the log producer stop.
+	_, _ = http.Get(ep + "/crash") // Will return an error.
+
+	// Wait for an arbitrary time to make sure that the log producer definitely
+	// exited.
+	<-time.After(200 * time.Millisecond)
+
+	// Stop the log producer. It should have already exited now.
+	err = c.StopLogProducer()
+	require.NoError(t, err)
+
+	assert.Equal(t, []string{
+		"ready\n",
+		"echo mlem\n",
+		"ready\n",
+		"echo mlem\n",
+		"echo mlem2\n",
+	}, g.Msgs)
+	assert.Nil(t, c.Terminate(ctx))
+}
+
 func TestContainerLogWithErrClosed(t *testing.T) {
 	t.Cleanup(func() {
 		config.Reset()

testdata/echoserver.go

@@ -29,13 +29,20 @@ func echoHandler(destination *os.File) http.HandlerFunc {
 	}
 }
 
+func crashHandler() http.HandlerFunc {
+	return func(w http.ResponseWriter, _ *http.Request) {
+		panic("manual panic via http")
+	}
+}
+
 // a simple server that will echo whatever is in the "echo" parameter to stdout
 // in the /stdout endpoint or to stderr in the /stderr endpoint
 func main() {
 	mux := http.NewServeMux()
 	mux.HandleFunc("/stdout", echoHandler(os.Stdout))
 	mux.HandleFunc("/stderr", echoHandler(os.Stderr))
 	mux.HandleFunc("/env", envHandler())
+	mux.HandleFunc("/crash", crashHandler())
 
 	ln, err := net.Listen("tcp", ":8080")
 	if err != nil {