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:
+ requests.post(self.intake_url, data=f, headers={'content-type': 'application/x-ndjson'})
+
+ # 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)

tests/system/test_setup_index_management.py

@@ -8,13 +8,15 @@
 
 INTEGRATION_TESTS = os.environ.get('INTEGRATION_TESTS', False)
 
+EVENT_NAMES = ('error', 'span', 'transaction', 'metric', 'profile')
+
 
 class IdxMgmt(object):
 
  def __init__(self, client, index):
  self._client = client
  self._index = index
- self._event_indices = ["{}-{}".format(self._index, e) for e in ['error', 'span', 'transaction', 'metric']]
+ self._event_indices = ["{}-{}".format(self._index, e) for e in EVENT_NAMES]
  self.default_policy = "apm-rollover-30-days"
 
  def indices(self):
@@ -40,7 +42,7 @@ def assert_template(self, loaded=1):
  s, i, l = 'settings', 'index', 'lifecycle'
  assert l not in resp[self._index][s][i]
 
- def assert_event_template(self, loaded=4, with_ilm=True):
+ def assert_event_template(self, loaded=len(EVENT_NAMES), with_ilm=True):
  resp = self._client.indices.get_template(name=self._index + '*', ignore=[404])
 
  if self._index in resp:
@@ -60,7 +62,7 @@ def assert_event_template(self, loaded=4, with_ilm=True):
  assert t[l]['name'] is not None, t[l]
  assert t[l]['rollover_alias'] == idx, t[l]
 
- def assert_alias(self, loaded=4):
+ def assert_alias(self, loaded=len(EVENT_NAMES)):
  resp = self._client.transport.perform_request('GET', '/_alias/' + self._index + '*')
  if loaded == 0:
  return self.assert_empty(resp)
@@ -321,7 +323,7 @@ def test_ilm_loaded(self):
  max_timeout=5)
 
  self.idxmgmt.assert_event_template(with_ilm=True)
- self.idxmgmt.assert_alias(loaded=4)
+ self.idxmgmt.assert_alias()
  self.idxmgmt.assert_default_policy(loaded=True)
 
  # check out configured policy in apm-server.yml.j2