diff --git a/libbeat/autodiscover/providers/docker/docker_integration_test.go b/libbeat/autodiscover/providers/docker/docker_integration_test.go index bbb2bc979bc..898f3cd254c 100644 --- a/libbeat/autodiscover/providers/docker/docker_integration_test.go +++ b/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 } diff --git a/libbeat/tests/docker/docker.go b/libbeat/tests/docker/docker.go index 888347c5cc7..8bb5efadbfa 100644 --- a/libbeat/tests/docker/docker.go +++ b/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{