Fix autodiscover flaky tests (elastic#21242)

Use a tagged image instead of doing an untagged pull, this was pulling all
busybox tags, busybox:latest image used now is pre-cached.
Better control when the containers are started and stopped, run the
container detached.
Use container name instead of container image for conditions and templates,
better ensuring that we are testing against the proper container.
Move common code to helper function.

filebeat/tests/system/test_autodiscover.py

@@ -1,8 +1,10 @@
-import os
+import docker
 import filebeat
+import os
 import unittest
 
 from beat.beat import INTEGRATION_TESTS
+from contextlib import contextmanager
 
 
 class TestAutodiscover(filebeat.BaseTest):
@@ -16,89 +18,78 @@ def test_docker(self):
         """
         Test docker autodiscover starts input
         """
-        import docker
-        docker_client = docker.from_env()
-
-        self.render_config_template(
-            inputs=False,
-            autodiscover={
-                'docker': {
-                    'cleanup_timeout': '0s',
-                    'templates': '''
-                      - condition:
-                          equals.docker.container.image: busybox
-                        config:
-                          - type: log
-                            paths:
-                              - %s/${data.docker.container.image}.log
-                    ''' % self.working_dir,
+        with self.container_running() as container:
+            self.render_config_template(
+                inputs=False,
+                autodiscover={
+                    'docker': {
+                        'cleanup_timeout': '0s',
+                        'templates': f'''
+                          - condition:
+                                equals.docker.container.name: {container.name}
+                            config:
+                              - type: log
+                                paths:
+                                  - %s/${{data.docker.container.name}}.log
+                        ''' % self.working_dir,
+                    },
                 },
-            },
-        )
+            )
 
-        with open(os.path.join(self.working_dir, 'busybox.log'), 'wb') as f:
-            f.write(b'Busybox output 1\n')
-
-        proc = self.start_beat()
-        docker_client.images.pull('busybox')
-        docker_client.containers.run('busybox', 'sleep 1')
+            proc = self.start_beat()
+            self._test(container)
 
-        self.wait_until(lambda: self.log_contains('Starting runner: input'))
         self.wait_until(lambda: self.log_contains('Stopping runner: input'))
-
-        output = self.read_output_json()
         proc.check_kill_and_wait()
 
-        # Check metadata is added
-        assert output[0]['message'] == 'Busybox output 1'
-        assert output[0]['container']['image']['name'] == 'busybox'
-        assert output[0]['docker']['container']['labels'] == {}
-        assert 'name' in output[0]['container']
-
-        self.assert_fields_are_documented(output[0])
-
     @unittest.skipIf(not INTEGRATION_TESTS or
                      os.getenv("TESTING_ENVIRONMENT") == "2x",
                      "integration test not available on 2.x")
     def test_default_settings(self):
         """
         Test docker autodiscover default config settings
         """
-        import docker
-        docker_client = docker.from_env()
-
-        self.render_config_template(
-            inputs=False,
-            autodiscover={
-                'docker': {
-                    'cleanup_timeout': '0s',
-                    'hints.enabled': 'true',
-                    'hints.default_config': '''
-                      type: log
-                      paths:
-                        - %s/${data.container.image}.log
-                    ''' % self.working_dir,
+        with self.container_running() as container:
+            self.render_config_template(
+                inputs=False,
+                autodiscover={
+                    'docker': {
+                        'cleanup_timeout': '0s',
+                        'hints.enabled': 'true',
+                        'hints.default_config': '''
+                          type: log
+                          paths:
+                            - %s/${data.container.name}.log
+                        ''' % self.working_dir,
+                    },
                 },
-            },
-        )
+            )
+            proc = self.start_beat()
+            self._test(container)
 
-        with open(os.path.join(self.working_dir, 'busybox.log'), 'wb') as f:
-            f.write(b'Busybox output 1\n')
+        self.wait_until(lambda: self.log_contains('Stopping runner: input'))
+        proc.check_kill_and_wait()
 
-        proc = self.start_beat()
-        docker_client.images.pull('busybox')
-        docker_client.containers.run('busybox', 'sleep 1')
+    def _test(self, container):
+        with open(os.path.join(self.working_dir, f'{container.name}.log'), 'wb') as f:
+            f.write(b'Busybox output 1\n')
 
         self.wait_until(lambda: self.log_contains('Starting runner: input'))
-        self.wait_until(lambda: self.log_contains('Stopping runner: input'))
+        self.wait_until(lambda: self.output_has(lines=1))
 
         output = self.read_output_json()
-        proc.check_kill_and_wait()
 
         # Check metadata is added
         assert output[0]['message'] == 'Busybox output 1'
-        assert output[0]['container']['image']['name'] == 'busybox'
-        assert output[0]['docker']['container']['labels'] == {}
+        assert output[0]['container']['name'] == container.name
+        assert output[0]['docker']['container']['labels'] == container.labels
         assert 'name' in output[0]['container']
 
         self.assert_fields_are_documented(output[0])
+
+    @contextmanager
+    def container_running(self, image_name='busybox:latest'):
+        docker_client = docker.from_env()
+        container = docker_client.containers.run(image_name, 'sleep 60', detach=True, remove=True)
+        yield container
+        container.remove(force=True)