regression-tests.dnsdist/test_Prometheus.py

   1 #!/usr/bin/env python
   2 import requests
   3 import subprocess
   4 from dnsdisttests import DNSDistTest
   5
   6 class TestPrometheus(DNSDistTest):
   7
   8     _webTimeout = 2.0
   9     _webServerPort = 8083
  10     _webServerBasicAuthPassword = 'secret'
  11     _webServerAPIKey = 'apisecret'
  12     _config_params = ['_testServerPort', '_webServerPort', '_webServerBasicAuthPassword', '_webServerAPIKey']
  13     _config_template = """
  14     newServer{address="127.0.0.1:%s"}
  15     webserver("127.0.0.1:%s", "%s", "%s")
  16     """
  17
  18     def checkPrometheusContentBasic(self, content):
  19         for line in content.splitlines():
  20             if line.startswith('# HELP'):
  21                 tokens = line.split(' ')
  22                 self.assertGreaterEqual(len(tokens), 4)
  23             elif line.startswith('# TYPE'):
  24                 tokens = line.split(' ')
  25                 self.assertEquals(len(tokens), 4)
  26                 self.assertIn(tokens[3], ['counter', 'gauge', 'histogram'])
  27             elif not line.startswith('#'):
  28                 tokens = line.split(' ')
  29                 self.assertEquals(len(tokens), 2)
  30                 if not line.startswith('dnsdist_'):
  31                     raise AssertionError('Expecting prometheus metric to be prefixed by \'dnsdist_\', got: "%s"' % (line))
  32
  33     def checkPrometheusContentPromtool(self, content):
  34         output = None
  35         try:
  36             testcmd = ['promtool', 'check', 'metrics']
  37             process = subprocess.Popen(testcmd, stdout=subprocess.PIPE, stdin=subprocess.PIPE, stderr=subprocess.STDOUT, close_fds=True)
  38             output = process.communicate(input=content)
  39         except subprocess.CalledProcessError as exc:
  40             raise AssertionError('%s failed (%d): %s' % (testcmd, process.returncode, process.output))
  41
  42         # commented out because promtool returns 3 because of the "_total" suffix warnings
  43         #if process.returncode != 0:
  44         #  raise AssertionError('%s failed (%d): %s' % (testcmd, process.returncode, output))
  45
  46         for line in output[0].splitlines():
  47             if line.endswith(b"should have \"_total\" suffix"):
  48                 continue
  49             raise AssertionError('%s returned an unexpected output. Faulty line is "%s", complete content is "%s"' % (testcmd, line, output))
  50
  51     def testMetrics(self):
  52         """
  53         Prometheus: Retrieve metrics
  54         """
  55         url = 'http://127.0.0.1:' + str(self._webServerPort) + '/metrics'
  56         r = requests.get(url, auth=('whatever', self._webServerBasicAuthPassword), timeout=self._webTimeout)
  57         self.assertTrue(r)
  58         self.assertEquals(r.status_code, 200)
  59         self.checkPrometheusContentBasic(r.text)
  60         self.checkPrometheusContentPromtool(r.content)