Update system tests (elastic#3014)

* expose elasticsearch and kibana
* include pipeline registration in started consideration
* check for index, pipeline, docs before destroying
* disable flaky test

Author: graphaelli

docker-compose.yml

@@ -23,11 +23,15 @@ services:
       kibana: { condition: service_healthy }
 
   elasticsearch:
+    ports:
+      - 9200:9200
     extends:
       file: ./_beats/testing/environments/${TESTING_ENVIRONMENT}.yml
       service: elasticsearch
 
   kibana:
+    ports:
+      - 5601:5601
     extends:
       file: ./_beats/testing/environments/${TESTING_ENVIRONMENT}.yml
       service: kibana

tests/system/apmserver.py

@@ -3,6 +3,7 @@
 import os
 import re
 import shutil
+import unittest
 from time import gmtime, strftime
 from urlparse import urlparse
 
@@ -14,8 +15,9 @@
 
 sys.path.append(os.path.join(os.path.dirname(__file__), '..',
                              '..', '_beats', 'libbeat', 'tests', 'system'))
-from beat.beat import TestCase, TimeoutError
+from beat.beat import INTEGRATION_TESTS, TestCase, TimeoutError
 
+integration_test = unittest.skipUnless(INTEGRATION_TESTS, "integration test")
 
 class BaseTest(TestCase):
     maxDiff = None
@@ -105,11 +107,26 @@ def setUp(self):
         self.apmserver_proc = self.start_beat(**self.start_args())
         self.wait_until_started()
 
+        # try make sure APM Server is fully up
+        cfg = self.config()
+        # pipeline registration is enabled by default and only happens if the output is elasticsearch
+        if not getattr(self, "register_pipeline_disabled", False) and \
+            cfg.get("elasticsearch_host") and \
+            cfg.get("register_pipeline_enabled") != "false" and cfg.get("register_pipeline_overwrite") != "false":
+            self.wait_until_pipelines_registered()
+
     def start_args(self):
         return {}
 
     def wait_until_started(self):
-        self.wait_until(lambda: self.log_contains("Starting apm-server"))
+        self.wait_until(lambda: self.log_contains("Starting apm-server"), name="apm-server started")
+
+    def wait_until_ilm_setup(self):
+        self.wait_until(lambda: self.log_contains("Finished index management setup."), name="ILM setup")
+
+    def wait_until_pipelines_registered(self):
+        self.wait_until(lambda: self.log_contains("Registered Ingest Pipelines successfully"),
+                                                  name="pipelines registered")
 
     def assert_no_logged_warnings(self, suppress=None):
         """
@@ -130,11 +147,13 @@ def assert_no_logged_warnings(self, suppress=None):
             log = re.sub(s, "", log)
         self.assertNotRegexpMatches(log, "ERR|WARN")
 
-    def request_intake(self, data="", url="", headers={'content-type': 'application/x-ndjson'}):
-        if url == "":
+    def request_intake(self, data=None, url=None, headers=None):
+        if not url:
             url = self.intake_url
-        if data == "":
+        if data is None:
             data = self.get_event_payload()
+        if headers is None:
+            headers = {'content-type': 'application/x-ndjson'}
         return requests.post(url, data=data, headers=headers)
 
 
@@ -181,20 +200,28 @@ def setUp(self):
         self.kibana_url = self.get_kibana_url()
 
         # Cleanup index and template first
-        self.es.indices.delete(index="apm*", ignore=[400, 404])
-        for idx in self.indices:
-            self.wait_until(lambda: not self.es.indices.exists(idx))
-
-        self.es.indices.delete_template(name="apm*", ignore=[400, 404])
-        for idx in self.indices:
-            self.wait_until(lambda: not self.es.indices.exists_template(idx))
+        assert all(idx.startswith("apm") for idx in self.indices), "not all indices prefixed with apm, cleanup assumption broken"
+        if self.es.indices.get("apm*"):
+            self.es.indices.delete(index="apm*", ignore=[400, 404])
+            for idx in self.indices:
+                self.wait_until(lambda: not self.es.indices.exists(idx), name="index {} to be deleted".format(idx))
+
+        if self.es.indices.get_template(name="apm*", ignore=[400, 404]):
+            self.es.indices.delete_template(name="apm*", ignore=[400, 404])
+            for idx in self.indices:
+                self.wait_until(lambda: not self.es.indices.exists_template(idx),
+                                name="index template {} to be deleted".format(idx))
 
         # truncate, don't delete agent configuration index since it's only created when kibana starts up
-        self.es.delete_by_query(self.index_acm, {"query": {"match_all": {}}},
-                                ignore_unavailable=True, wait_for_completion=True)
-        self.wait_until(lambda: self.es.count(index=self.index_acm, ignore_unavailable=True)["count"] == 0)
+        if self.es.count(index=self.index_acm, ignore_unavailable=True)["count"] > 0:
+            self.es.delete_by_query(self.index_acm, {"query": {"match_all": {}}},
+                                    ignore_unavailable=True, wait_for_completion=True)
+            self.wait_until(lambda: self.es.count(index=self.index_acm, ignore_unavailable=True)["count"] == 0,
+                            max_timeout=30, name="acm index {} to be empty".format(self.index_acm))
+
         # Cleanup pipelines
-        self.es.ingest.delete_pipeline(id="*")
+        if self.es.ingest.get_pipeline(ignore=[400, 404]):
+            self.es.ingest.delete_pipeline(id="*")
 
         super(ElasticTest, self).setUp()
 
@@ -219,7 +246,8 @@ def load_docs_with_template(self, data_path, url, endpoint, expected_events_coun
             lambda: (self.es.count(index=query_index, body={
                 "query": {"term": {"processor.name": endpoint}}}
             )['count'] == expected_events_count),
-            max_timeout=max_timeout
+            max_timeout=max_timeout,
+            name="{} documents to reach {}".format(endpoint, expected_events_count),
         )
 
     def check_backend_error_sourcemap(self, index, count=1):
@@ -311,7 +339,8 @@ def wait_for_sourcemaps(self, expected_ct=1):
         self.wait_until(
             lambda: (self.es.count(index=idx, body={
                 "query": {"term": {"processor.name": 'sourcemap'}}}
-            )['count'] == expected_ct)
+            )['count'] == expected_ct),
+            name="{} sourcemaps to ingest".format(expected_ct),
         )
 
     def check_rum_error_sourcemap(self, updated, expected_err=None, count=1):

tests/system/config/apm-server.yml.j2

@@ -70,7 +70,7 @@ apm-server:
   register.ingest.pipeline.enabled: {{ register_pipeline_enabled }}
   {% endif %}
   {% if register_pipeline_overwrite %}
-  register.ingest.pipeline.overwrite: {{ register_pipeline_overwrite  }}
+  register.ingest.pipeline.overwrite: {{ register_pipeline_overwrite }}
   {% endif %}
 
   {% if instrumentation_enabled %}

tests/system/test_access.py

@@ -9,7 +9,7 @@
 
 from nose.tools import raises
 from requests.exceptions import SSLError, ChunkedEncodingError
-from beat.beat import INTEGRATION_TESTS, TimeoutError
+from apmserver import TimeoutError, integration_test
 from requests.packages.urllib3.exceptions import SubjectAltNameWarning
 requests.packages.urllib3.disable_warnings(SubjectAltNameWarning)
 
@@ -56,7 +56,7 @@ def oauth(v):
         assert r.status_code == 401, r.status_code
 
 
-@unittest.skipUnless(INTEGRATION_TESTS, "integration test")
+@integration_test
 class TestSecureServerBaseTest(ServerSetUpBaseTest):
     @classmethod
     def setUpClass(cls):
@@ -145,6 +145,7 @@ def setUp(self):
         super(TestSecureServerBaseTest, self).setUp()
 
 
+@integration_test
 class TestSSLEnabledNoClientVerificationTest(TestSecureServerBaseTest):
     def ssl_overrides(self):
         return {"ssl_client_authentication": "none"}
@@ -163,6 +164,7 @@ def test_https_server_validation_fails(self):
         assert r.status_code == 202, r.status_code
 
 
+@integration_test
 class TestSSLEnabledOptionalClientVerificationTest(TestSecureServerBaseTest):
     # no ssl_overrides necessary as `optional` is default
 
@@ -183,6 +185,7 @@ def test_https_self_signed_cert(self):
         assert r.status_code == 202, r.status_code
 
 
+@integration_test
 class TestSSLEnabledOptionalClientVerificationWithCATest(TestSecureServerBaseTest):
     def ssl_overrides(self):
         return {"ssl_certificate_authorities": self.ca_cert}
@@ -205,6 +208,7 @@ def test_https_auth_cert_ok(self):
         assert r.status_code == 202, r.status_code
 
 
+@integration_test
 class TestSSLEnabledRequiredClientVerificationTest(TestSecureServerBaseTest):
     def ssl_overrides(self):
         return {"ssl_client_authentication": "required",
@@ -225,6 +229,7 @@ def test_https_auth_cert_ok(self):
         assert r.status_code == 202, r.status_code
 
 
+@integration_test
 class TestSSLDefaultSupportedProcotolsTest(TestSecureServerBaseTest):
     def ssl_overrides(self):
         return {"ssl_certificate_authorities": self.ca_cert}
@@ -240,6 +245,7 @@ def test_tls_v1_2(self):
         self.ssl_connect()
 
 
+@integration_test
 class TestSSLSupportedProcotolsTest(TestSecureServerBaseTest):
     def ssl_overrides(self):
         return {"ssl_supported_protocols": ["TLSv1.2"],
@@ -253,6 +259,7 @@ def test_tls_v1_2(self):
         self.ssl_connect()
 
 
+@integration_test
 class TestSSLSupportedCiphersTest(TestSecureServerBaseTest):
     def ssl_overrides(self):
         return {"ssl_cipher_suites": ['ECDHE-RSA-AES-128-GCM-SHA256'],

tests/system/test_export.py

@@ -2,10 +2,15 @@
 import os
 import json
 import shutil
-from apmserver import SubCommandTest
+from apmserver import SubCommandTest, integration_test
 
 
-class ExportConfigDefaultTest(SubCommandTest):
+class ExportCommandTest(SubCommandTest):
+    register_pipeline_disabled = True
+
+
+@integration_test
+class ExportConfigDefaultTest(ExportCommandTest):
     """
     Test export config subcommand.
     """
@@ -44,7 +49,8 @@ def test_export_config(self):
             }, config["setup"])
 
 
-class ExportConfigTest(SubCommandTest):
+@integration_test
+class ExportConfigTest(ExportCommandTest):
     """
     Test export config subcommand.
     """
@@ -86,7 +92,8 @@ def test_export_config(self):
             }, config["setup"])
 
 
-class TestExportTemplate(SubCommandTest):
+@integration_test
+class TestExportTemplate(ExportCommandTest):
     """
     Test export template
     """
@@ -118,7 +125,8 @@ def test_export_template_to_file(self):
         assert template['order'] == 1
 
 
-class TestExportILMPolicy(SubCommandTest):
+@integration_test
+class TestExportILMPolicy(ExportCommandTest):
     """
     Test export ilm-policy
     """
@@ -153,6 +161,7 @@ def test_export_ilm_policy_to_files(self):
                 assert "delete" not in policy["policy"]["phases"]
 
 
+@integration_test
 class TestExportILMPolicyILMDisabled(TestExportILMPolicy):
     """
     Test export ilm-policy independent of ILM enabled state