import requests
import subprocess
import unittest
-from dnsdisttests import DNSDistTest
+from dnsdisttests import DNSDistTest, pickAvailablePort
@unittest.skipIf('SKIP_PROMETHEUS_TESTS' in os.environ, 'Prometheus tests are disabled')
class TestPrometheus(DNSDistTest):
_webTimeout = 2.0
- _webServerPort = 8083
+ _webServerPort = pickAvailablePort()
_webServerBasicAuthPassword = 'secret'
+ _webServerBasicAuthPasswordHashed = '$scrypt$ln=10,p=1,r=8$6DKLnvUYEeXWh3JNOd3iwg==$kSrhdHaRbZ7R74q3lGBqO1xetgxRxhmWzYJ2Qvfm7JM='
_webServerAPIKey = 'apisecret'
- _config_params = ['_testServerPort', '_webServerPort', '_webServerBasicAuthPassword', '_webServerAPIKey']
+ _webServerAPIKeyHashed = '$scrypt$ln=10,p=1,r=8$9v8JxDfzQVyTpBkTbkUqYg==$bDQzAOHeK1G9UvTPypNhrX48w974ZXbFPtRKS34+aso='
+ _config_params = ['_testServerPort', '_webServerPort', '_webServerBasicAuthPasswordHashed', '_webServerAPIKeyHashed']
_config_template = """
newServer{address="127.0.0.1:%s"}
- webserver("127.0.0.1:%s", "%s", "%s")
+ webserver("127.0.0.1:%s")
+ setWebserverConfig({password="%s", apiKey="%s"})
+ pc = newPacketCache(100, {maxTTL=86400, minTTL=1})
+ getPool(""):setCache(pc)
+
+ -- test custom metrics as well
+ declareMetric('custom-metric1', 'counter', 'Custom counter')
+ incMetric('custom-metric1')
+ declareMetric('custom-metric2', 'gauge', 'Custom gauge')
+ -- and custom names
+ declareMetric('custom-metric3', 'counter', 'Custom counter', 'custom_prometheus_name')
"""
def checkPrometheusContentBasic(self, content):
self.assertGreaterEqual(len(tokens), 4)
elif line.startswith('# TYPE'):
tokens = line.split(' ')
- self.assertEquals(len(tokens), 4)
+ self.assertEqual(len(tokens), 4)
self.assertIn(tokens[3], ['counter', 'gauge', 'histogram'])
elif not line.startswith('#'):
tokens = line.split(' ')
- self.assertEquals(len(tokens), 2)
- if not line.startswith('dnsdist_'):
+ self.assertEqual(len(tokens), 2)
+ if not line.startswith('dnsdist_') and not line.startswith('custom_prometheus_name'):
raise AssertionError('Expecting prometheus metric to be prefixed by \'dnsdist_\', got: "%s"' % (line))
+ def checkMetric(self, content, name, expectedType, expectedValue):
+ typeFound = False
+ helpFound = False
+ valueFound = False
+ for line in content.splitlines():
+ if name in str(line):
+ tokens = line.split(' ')
+ if line.startswith('# HELP'):
+ self.assertGreaterEqual(len(tokens), 4)
+ if tokens[2] == name:
+ helpFound = True
+ elif line.startswith('# TYPE'):
+ self.assertEqual(len(tokens), 4)
+ if tokens[2] == name:
+ typeFound = True
+ self.assertEqual(tokens[3], expectedType)
+ elif not line.startswith('#'):
+ self.assertEqual(len(tokens), 2)
+ if tokens[0] == name:
+ valueFound = True
+ self.assertEqual(int(tokens[1]), expectedValue)
+
+ self.assertTrue(typeFound)
+ self.assertTrue(helpFound)
+ self.assertTrue(valueFound)
+
def checkPrometheusContentPromtool(self, content):
output = None
try:
url = 'http://127.0.0.1:' + str(self._webServerPort) + '/metrics'
r = requests.get(url, auth=('whatever', self._webServerBasicAuthPassword), timeout=self._webTimeout)
self.assertTrue(r)
- self.assertEquals(r.status_code, 200)
+ self.assertEqual(r.status_code, 200)
self.checkPrometheusContentBasic(r.text)
self.checkPrometheusContentPromtool(r.content)
+ self.checkMetric(r.text, 'dnsdist_custom_metric1', 'counter', 1)
+ self.checkMetric(r.text, 'dnsdist_custom_metric2', 'gauge', 0)
+ self.checkMetric(r.text, 'custom_prometheus_name', 'counter', 0)