ceph/qa/tasks/mgr/test_prometheus.py

   1 from tasks.mgr.mgr_test_case import MgrTestCase
   2
   3 import json
   4 import logging
   5 import requests
   6
   7
   8 log = logging.getLogger(__name__)
   9
  10
  11 class TestPrometheus(MgrTestCase):
  12     MGRS_REQUIRED = 3
  13
  14     def setUp(self):
  15         super(TestPrometheus, self).setUp()
  16         self.setup_mgrs()
  17
  18     def test_file_sd_command(self):
  19         self._assign_ports("prometheus", "server_port")
  20         self._load_module("prometheus")
  21
  22         result = json.loads(self.mgr_cluster.mon_manager.raw_cluster_cmd(
  23             "prometheus", "file_sd_config"))
  24         mgr_map = self.mgr_cluster.get_mgr_map()
  25         self.assertEqual(len(result[0]['targets']), len(mgr_map['standbys']) + 1)
  26
  27
  28
  29     def test_standby(self):
  30         self._assign_ports("prometheus", "server_port")
  31         self._load_module("prometheus")
  32
  33         original_active = self.mgr_cluster.get_active_id()
  34
  35         original_uri = self._get_uri("prometheus")
  36         log.info("Originally running at {0}".format(original_uri))
  37
  38         self.mgr_cluster.mgr_fail(original_active)
  39
  40         failed_over_uri = self._get_uri("prometheus")
  41         log.info("After failover running at {0}".format(failed_over_uri))
  42
  43         self.assertNotEqual(original_uri, failed_over_uri)
  44
  45         # The original active daemon should have come back up as a standby
  46         # and serve some html under "/" and an empty answer under /metrics
  47         r = requests.get(original_uri, allow_redirects=False)
  48         self.assertEqual(r.status_code, 200)
  49         r = requests.get(original_uri + "metrics", allow_redirects=False)
  50         self.assertEqual(r.status_code, 200)
  51         self.assertEqual(r.headers["content-type"], "text/plain;charset=utf-8")
  52
  53     def test_urls(self):
  54         self._assign_ports("prometheus", "server_port")
  55         self._load_module("prometheus")
  56
  57         base_uri = self._get_uri("prometheus")
  58
  59         # This is a very simple smoke test to check that the module can
  60         # give us a 200 response to requests.  We're not testing that
  61         # the content is correct or even renders!
  62
  63         urls = [
  64             "/",
  65             "/metrics"
  66         ]
  67
  68         failures = []
  69
  70         for url in urls:
  71             r = requests.get(base_uri + url, allow_redirects=False)
  72             if r.status_code != 200:
  73                 failures.append(url)
  74
  75             log.info("{0}: {1} ({2} bytes)".format(
  76                 url, r.status_code, len(r.content)
  77             ))
  78
  79         self.assertListEqual(failures, [])