regression-tests.dnsdist/test_Prometheus.py

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