Merge pull request #584 from docker/aci_stop_follow_logs

ACI : when following logs (actually polling logs), stop polling when the container is not running.

aci/aci.go

@@ -259,6 +259,8 @@ func getACIContainerLogs(ctx context.Context, aciContext store.AciContext, conta
 
 func streamLogs(ctx context.Context, aciContext store.AciContext, containerGroupName, containerName string, req containers.LogsRequest) error {
 	numLines := 0
+	previousLogLines := ""
+	firstDisplay := true // optimization to exit sooner in cases like docker run hello-world, do not wait another 2 secs.
 	for {
 		select {
 		case <-ctx.Done():
@@ -285,6 +287,12 @@ func streamLogs(ctx context.Context, aciContext store.AciContext, containerGroup
 				fmt.Fprintln(req.Writer, logLines[i])
 			}
 
+			if (firstDisplay || previousLogLines == logs) && !isContainerRunning(ctx, aciContext, containerGroupName, containerName) {
+				return nil
+			}
+			firstDisplay = false
+			previousLogLines = logs
+
 			select {
 			case <-ctx.Done():
 				return nil
@@ -294,6 +302,21 @@ func streamLogs(ctx context.Context, aciContext store.AciContext, containerGroup
 	}
 }
 
+func isContainerRunning(ctx context.Context, aciContext store.AciContext, containerGroupName, containerName string) bool {
+	group, err := getACIContainerGroup(ctx, aciContext, containerGroupName)
+	if err != nil {
+		return false // group has disappeared
+	}
+	for _, container := range *group.Containers {
+		if *container.Name == containerName {
+			if convert.GetStatus(container, group) == convert.StatusRunning {
+				return true
+			}
+		}
+	}
+	return false
+}
+
 func getBacktrackLines(lines []string, terminalWidth int) int {
 	if terminalWidth == 0 { // no terminal width has been set, do not divide by zero
 		return len(lines)

aci/backend.go

@@ -48,7 +48,6 @@ const (
 	singleContainerTag        = "docker-single-container"
 	composeContainerTag       = "docker-compose-application"
 	composeContainerSeparator = "_"
-	statusRunning             = "Running"
 )
 
 // ContextParams options for creating ACI context
@@ -183,7 +182,7 @@ func getContainerID(group containerinstance.ContainerGroup, container containeri
 }
 
 func isContainerVisible(container containerinstance.Container, group containerinstance.ContainerGroup, showAll bool) bool {
-	return *container.Name == convert.ComposeDNSSidecarName || (!showAll && convert.GetStatus(container, group) != statusRunning)
+	return *container.Name == convert.ComposeDNSSidecarName || (!showAll && convert.GetStatus(container, group) != convert.StatusRunning)
 }
 
 func (cs *aciContainerService) Run(ctx context.Context, r containers.ContainerConfig) error {
@@ -349,7 +348,7 @@ func (cs *aciContainerService) Delete(ctx context.Context, containerID string, r
 		for _, container := range *cg.Containers {
 			status := convert.GetStatus(container, cg)
 
-			if status == statusRunning {
+			if status == convert.StatusRunning {
 				return errdefs.ErrForbidden
 			}
 		}

aci/convert/convert.go

@@ -40,6 +40,8 @@ import (
 )
 
 const (
+	// StatusRunning name of the ACI running status
+	StatusRunning = "Running"
 	// ComposeDNSSidecarName name of the dns sidecar container
 	ComposeDNSSidecarName = "aci--dns--sidecar"
 	dnsSidecarImage       = "busybox:1.31.1"
@@ -391,7 +393,7 @@ func bytesToGb(b types.UnitBytes) float64 {
 // ContainerGroupToServiceStatus convert from an ACI container definition to service status
 func ContainerGroupToServiceStatus(containerID string, group containerinstance.ContainerGroup, container containerinstance.Container) compose.ServiceStatus {
 	var replicas = 1
-	if GetStatus(container, group) != "Running" {
+	if GetStatus(container, group) != StatusRunning {
 		replicas = 0
 	}
 	return compose.ServiceStatus{

tests/aci-e2e/e2e-aci_test.go

@@ -31,6 +31,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/docker/compose-cli/aci/convert"
+
 	"gotest.tools/v3/assert"
 	is "gotest.tools/v3/assert/cmp"
 	"gotest.tools/v3/icmd"
@@ -294,6 +296,7 @@ func TestContainerRunAttached(t *testing.T) {
 
 	container = "test-container"
 
+	var followLogsProcess *icmd.Result
 	t.Run("run attached limits", func(t *testing.T) {
 		cmd := c.NewDockerCmd(
 			"run",
@@ -303,7 +306,7 @@ func TestContainerRunAttached(t *testing.T) {
 			"-p", "80:80",
 			"nginx",
 		)
-		runRes := icmd.StartCmd(cmd)
+		followLogsProcess = icmd.StartCmd(cmd)
 
 		checkRunning := func(t poll.LogT) poll.Result {
 			res := c.RunDockerOrExitError("inspect", container)
@@ -330,7 +333,7 @@ func TestContainerRunAttached(t *testing.T) {
 		assert.Equal(t, port.HostPort, uint32(80))
 		endpoint = fmt.Sprintf("http://%s:%d", port.HostIP, port.HostPort)
 
-		assert.Assert(t, !strings.Contains(runRes.Stdout(), "/test"))
+		assert.Assert(t, !strings.Contains(followLogsProcess.Stdout(), "/test"))
 		checkRequest := func(t poll.LogT) poll.Result {
 			r, _ := http.Get(endpoint + "/test")
 			if r != nil && r.StatusCode == http.StatusNotFound {
@@ -341,7 +344,7 @@ func TestContainerRunAttached(t *testing.T) {
 		poll.WaitOn(t, checkRequest, poll.WithDelay(1*time.Second), poll.WithTimeout(60*time.Second))
 
 		checkLog := func(t poll.LogT) poll.Result {
-			if strings.Contains(runRes.Stdout(), "/test") {
+			if strings.Contains(followLogsProcess.Stdout(), "/test") {
 				return poll.Success()
 			}
 			return poll.Continue("waiting for logs to contain /test")
@@ -360,6 +363,13 @@ func TestContainerRunAttached(t *testing.T) {
 	t.Run("stop container", func(t *testing.T) {
 		res := c.RunDockerCmd("stop", container)
 		res.Assert(t, icmd.Expected{Out: container})
+		waitForStatus(t, c, container, "Terminated", "Node Stopped")
+	})
+
+	t.Run("check we stoppped following logs", func(t *testing.T) {
+		// nolint errcheck
+		followLogsStopped := waitWithTimeout(func() { followLogsProcess.Cmd.Process.Wait() }, 10*time.Second)
+		assert.NilError(t, followLogsStopped, "Follow logs process did not stop after container is stopped")
 	})
 
 	t.Run("ps stopped container with --all", func(t *testing.T) {
@@ -372,14 +382,14 @@ func TestContainerRunAttached(t *testing.T) {
 		assert.Assert(t, is.Len(out, 2))
 	})
 
-	t.Run("start container", func(t *testing.T) {
+	t.Run("restart container", func(t *testing.T) {
 		res := c.RunDockerCmd("start", container)
 		res.Assert(t, icmd.Expected{Out: container})
-		waitForStatus(t, c, container, "Running")
+		waitForStatus(t, c, container, convert.StatusRunning)
 	})
 
-	t.Run("rm stopped container", func(t *testing.T) {
-		res := c.RunDockerCmd("stop", container)
+	t.Run("kill & rm stopped container", func(t *testing.T) {
+		res := c.RunDockerCmd("kill", container)
 		res.Assert(t, icmd.Expected{Out: container})
 		waitForStatus(t, c, container, "Terminated", "Node Stopped")
 
@@ -675,3 +685,17 @@ func waitForStatus(t *testing.T, c *E2eCLI, containerID string, statuses ...stri
 
 	poll.WaitOn(t, checkStopped, poll.WithDelay(5*time.Second), poll.WithTimeout(90*time.Second))
 }
+
+func waitWithTimeout(blockingCall func(), timeout time.Duration) error {
+	c := make(chan struct{})
+	go func() {
+		defer close(c)
+		blockingCall()
+	}()
+	select {
+	case <-c:
+		return nil
+	case <-time.After(timeout):
+		return fmt.Errorf("Timed out after %s", timeout)
+	}
+}