tests/system: add system tests for profiling

tests/system/apmserver.py

@@ -37,6 +37,7 @@ def setUpClass(cls):
         cls.index_span = "apm-{}-span".format(cls.apm_version)
         cls.index_metric = "apm-{}-metric".format(cls.apm_version)
         cls.index_smap = "apm-{}-sourcemap".format(cls.apm_version)
+        cls.index_profile = "apm-{}-profile".format(cls.apm_version)
         cls.index_acm = ".apm-agent-configuration"
         cls.indices = [cls.index_onboarding, cls.index_error,
                        cls.index_transaction, cls.index_span, cls.index_metric, cls.index_smap]

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

@@ -73,6 +73,25 @@ apm-server:
   register.ingest.pipeline.overwrite: {{ register_pipeline_overwrite  }}
   {% endif %}
 
+  {% if instrumentation_enabled %}
+  instrumentation.enabled: {{ instrumentation_enabled }}
+  {% endif %}
+  {% if profiling_cpu_enabled %}
+  instrumentation.profiling.cpu.enabled: {{ profiling_cpu_enabled }}
+  {% endif %}
+  {% if profiling_cpu_interval %}
+  instrumentation.profiling.cpu.interval: {{ profiling_cpu_interval }}
+  {% endif %}
+  {% if profiling_cpu_duration %}
+  instrumentation.profiling.cpu.duration: {{ profiling_cpu_duration }}
+  {% endif %}
+  {% if profiling_heap_enabled %}
+  instrumentation.profiling.heap.enabled: {{ profiling_heap_enabled }}
+  {% endif %}
+  {% if profiling_heap_interval %}
+  instrumentation.profiling.heap.interval: {{ profiling_heap_interval }}
+  {% endif %}
+
   {% if mode %}
   mode: {{ mode }}
   {% endif %}

tests/system/test_integration.py

@@ -629,6 +629,80 @@ def test_metric_doc(self):
         assert expected_type == actual_type, "want: {}, got: {}".format(expected_type, actual_type)
 
 
+class ProfileIntegrationTest(ElasticTest):
+    def metric_fields(self):
+        metric_fields = set()
+        rs = self.es.search(index=self.index_profile)
+        for hit in rs["hits"]["hits"]:
+            profile = hit["_source"]["profile"]
+            metric_fields.update((k for (k, v) in profile.items() if type(v) is int))
+        return metric_fields
+
+    def wait_for_profile(self):
+        def cond():
+            self.es.indices.refresh(index=self.index_profile)
+            response = self.es.count(index=self.index_profile, body={"query": {"term": {"processor.name": "profile"}}})
+            return response['count'] != 0
+        self.wait_until(cond, max_timeout=10)
+
+
+class CPUProfileIntegrationTest(ProfileIntegrationTest):
+    config_overrides = {
+        "instrumentation_enabled": "true",
+        "profiling_cpu_enabled": "true",
+        "profiling_cpu_interval": "1s",
+        "profiling_cpu_duration": "5s",
+    }
+
+    @unittest.skipUnless(INTEGRATION_TESTS, "integration test")
+    def test_self_profiling(self):
+        """CPU profiling enabled"""
+
+        import requests
+        def create_load():
+            payload_path = self.get_payload_path("transactions_spans.ndjson")
+            with open(payload_path) as f:
+                r = requests.post(self.intake_url, data=f, headers={'content-type': 'application/x-ndjson'})
+                assert r.status_code == 202, r.status_code
+
+        # Wait for profiling to begin, and then start sending data
+        # to the server to create some CPU load.
+        from datetime import datetime, timedelta
+        time.sleep(1)
+        start = datetime.now()
+        while datetime.now()-start < timedelta(seconds=5):
+            create_load()
+        self.wait_for_profile()
+
+        expected_metric_fields = set([u"cpu.ns", u"samples.count", u"duration"])
+        metric_fields = self.metric_fields()
+        self.assertEqual(metric_fields, expected_metric_fields)
+
+
+class HeapProfileIntegrationTest(ProfileIntegrationTest):
+    config_overrides = {
+        "instrumentation_enabled": "true",
+        "profiling_heap_enabled": "true",
+        "profiling_heap_interval": "1s",
+    }
+
+    @unittest.skipUnless(INTEGRATION_TESTS, "integration test")
+    def test_self_profiling(self):
+        """Heap profiling enabled"""
+
+        time.sleep(1)
+        self.wait_for_profile()
+
+        expected_metric_fields = set([
+            u"alloc_objects.count",
+            u"inuse_objects.count",
+            u"alloc_space.bytes",
+            u"inuse_space.bytes",
+        ])
+        metric_fields = self.metric_fields()
+        self.assertEqual(metric_fields, expected_metric_fields)
+
+
 class ExperimentalBaseTest(ElasticTest):
     def check_experimental_key_indexed(self, experimental):
         self.wait_until(lambda: self.log_contains("Registered Ingest Pipelines successfully"), max_timeout=10)