Fix TestDockerStart flaky test (elastic#21681)

Some changes are done to give more resilience to the test:
* Wait till image pull is finished, and retry in case of failure.
* Checked events are filtered by container id instead of image name,
  so tests are not affected by other containers that may be running
  in the system.
* Check timeout is for all events now, instead of being reset after an
  event is received.
* Container is removed after test is finished.

(cherry picked from commit a79dddc)

libbeat/autodiscover/providers/docker/docker_integration_test.go

@@ -36,8 +36,6 @@ import (
 
 // Test docker start emits an autodiscover event
 func TestDockerStart(t *testing.T) {
-	t.Skip("#20360 Flaky TestDockerStart skipped")
-
 	log := logp.NewLogger("docker")
 
 	d, err := dk.NewClient()
@@ -70,15 +68,17 @@ func TestDockerStart(t *testing.T) {
 	// Start
 	cmd := []string{"echo", "Hi!"}
 	labels := map[string]string{"label": "foo", "label.child": "bar"}
-	ID, err := d.ContainerStart("busybox", cmd, labels)
+	ID, err := d.ContainerStart("busybox:latest", cmd, labels)
 	if err != nil {
 		t.Fatal(err)
 	}
-	checkEvent(t, listener, true)
+	defer d.ContainerRemove(ID)
+
+	checkEvent(t, listener, ID, true)
 
 	// Kill
 	d.ContainerKill(ID)
-	checkEvent(t, listener, false)
+	checkEvent(t, listener, ID, false)
 }
 
 func getValue(e bus.Event, key string) interface{} {
@@ -89,12 +89,13 @@ func getValue(e bus.Event, key string) interface{} {
 	return val
 }
 
-func checkEvent(t *testing.T, listener bus.Listener, start bool) {
+func checkEvent(t *testing.T, listener bus.Listener, id string, start bool) {
+	timeout := time.After(60 * time.Second)
 	for {
 		select {
 		case e := <-listener.Events():
 			// Ignore any other container
-			if getValue(e, "docker.container.image") != "busybox" {
+			if getValue(e, "container.id") != id {
 				continue
 			}
 			if start {
@@ -104,7 +105,7 @@ func checkEvent(t *testing.T, listener bus.Listener, start bool) {
 				assert.Equal(t, getValue(e, "stop"), true)
 				assert.Nil(t, getValue(e, "start"))
 			}
-			assert.Equal(t, getValue(e, "container.image.name"), "busybox")
+			assert.Equal(t, getValue(e, "container.image.name"), "busybox:latest")
 			// labels.dedot=true by default
 			assert.Equal(t,
 				common.MapStr{
@@ -122,8 +123,7 @@ func checkEvent(t *testing.T, listener bus.Listener, start bool) {
 			assert.Equal(t, getValue(e, "docker.container.name"), getValue(e, "meta.container.name"))
 			assert.Equal(t, getValue(e, "docker.container.image"), getValue(e, "meta.container.image.name"))
 			return
-
-		case <-time.After(10 * time.Second):
+		case <-timeout:
 			t.Fatal("Timeout waiting for provider events")
 			return
 		}

libbeat/tests/docker/docker.go

@@ -19,6 +19,8 @@ package docker
 
 import (
 	"context"
+	"io"
+	"io/ioutil"
 
 	"github.com/pkg/errors"
 
@@ -42,13 +44,12 @@ func NewClient() (Client, error) {
 
 // ContainerStart pulls and starts the given container
 func (c Client) ContainerStart(image string, cmd []string, labels map[string]string) (string, error) {
-	ctx := context.Background()
-	respBody, err := c.cli.ImagePull(ctx, image, types.ImagePullOptions{})
+	err := c.imagePull(image)
 	if err != nil {
-		return "", errors.Wrapf(err, "pullling image %s", image)
+		return "", err
 	}
-	defer respBody.Close()
 
+	ctx := context.Background()
 	resp, err := c.cli.ContainerCreate(ctx, &container.Config{
 		Image:  image,
 		Cmd:    cmd,
@@ -65,6 +66,36 @@ func (c Client) ContainerStart(image string, cmd []string, labels map[string]str
 	return resp.ID, nil
 }
 
+// imagePull pulls an image
+func (c Client) imagePull(image string) (err error) {
+	ctx := context.Background()
+	_, _, err = c.cli.ImageInspectWithRaw(ctx, image)
+	if err == nil {
+		// Image already available, do nothing
+		return nil
+	}
+	for retry := 0; retry < 3; retry++ {
+		err = func() error {
+			respBody, err := c.cli.ImagePull(ctx, image, types.ImagePullOptions{})
+			if err != nil {
+				return errors.Wrapf(err, "pullling image %s", image)
+			}
+			defer respBody.Close()
+
+			// Read all the response, to be sure that the pull has finished before returning.
+			_, err = io.Copy(ioutil.Discard, respBody)
+			if err != nil {
+				return errors.Wrapf(err, "reading response for image %s", image)
+			}
+			return nil
+		}()
+		if err == nil {
+			break
+		}
+	}
+	return
+}
+
 // ContainerWait waits for a container to finish
 func (c Client) ContainerWait(ID string) error {
 	ctx := context.Background()
@@ -89,7 +120,7 @@ func (c Client) ContainerKill(ID string) error {
 	return c.cli.ContainerKill(ctx, ID, "KILL")
 }
 
-// ContainerRemove kills and removed the given container
+// ContainerRemove kills and removes the given container
 func (c Client) ContainerRemove(ID string) error {
 	ctx := context.Background()
 	return c.cli.ContainerRemove(ctx, ID, types.ContainerRemoveOptions{