# Copyright 2013 Hewlett-Packard Development Company, L.P. # Copyright 2014 OpenStack Foundation # Copyright 2018 Red Hat, Inc. # # Licensed under the Apache License, Version 2.0 (the "License"); you may # not use this file except in compliance with the License. You may obtain # a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the # License for the specific language governing permissions and limitations # under the License. import itertools import os import pprint import select import socket import threading import time import fixtures from keystoneauth1 import exceptions import prometheus_client from requests import exceptions as rexceptions import testtools.content from openstack.tests.unit import base class StatsdFixture(fixtures.Fixture): def _setUp(self): self.running = True self.thread = threading.Thread(target=self.run) self.thread.daemon = True self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) self.sock.bind(('', 0)) self.port = self.sock.getsockname()[1] self.wake_read, self.wake_write = os.pipe() self.stats = [] self.thread.start() self.addCleanup(self._cleanup) def run(self): while self.running: poll = select.poll() poll.register(self.sock, select.POLLIN) poll.register(self.wake_read, select.POLLIN) ret = poll.poll() for fd, event in ret: if fd == self.sock.fileno(): data = self.sock.recvfrom(1024) if not data: return self.stats.append(data[0]) if fd == self.wake_read: return def _cleanup(self): self.running = False os.write(self.wake_write, b'1\n') self.thread.join() class TestStats(base.TestCase): def setUp(self): self.statsd = StatsdFixture() self.useFixture(self.statsd) # note, use 127.0.0.1 rather than localhost to avoid getting ipv6 # see: https://github.com/jsocol/pystatsd/issues/61 self.useFixture( fixtures.EnvironmentVariable('STATSD_HOST', '127.0.0.1') ) self.useFixture( fixtures.EnvironmentVariable('STATSD_PORT', str(self.statsd.port)) ) self.add_info_on_exception('statsd_content', self.statsd.stats) # Set up the above things before the super setup so that we have the # environment variables set when the Connection is created. super().setUp() self._registry = prometheus_client.CollectorRegistry() self.cloud.config._collector_registry = self._registry self.addOnException(self._add_prometheus_samples) def _add_prometheus_samples(self, exc_info): samples = [] for metric in self._registry.collect(): for s in metric.samples: samples.append(s) self.addDetail( 'prometheus_samples', testtools.content.text_content(pprint.pformat(samples)), ) def assert_reported_stat(self, key, value=None, kind=None): """Check statsd output Check statsd return values. A ``value`` should specify a ``kind``, however a ``kind`` may be specified without a ``value`` for a generic match. Leave both empty to just check for key presence. :arg str key: The statsd key :arg str value: The expected value of the metric ``key`` :arg str kind: The expected type of the metric ``key`` For example - ``c`` counter - ``g`` gauge - ``ms`` timing - ``s`` set Note that for ``ms`` type, you are expressing a maximum value, not an exact value. This is to avoid flakey tests. """ self.assertIsNotNone(self.statsd) if value: self.assertNotEqual(kind, None) start = time.time() while time.time() < (start + 1): # Note our fake statsd just queues up results in a queue. # We just keep going through them until we find one that # matches, or fail out. If statsd pipelines are used, # large single packets are sent with stats separated by # newlines; thus we first flatten the stats out into # single entries. stats = itertools.chain.from_iterable( [s.decode('utf-8').split('\n') for s in self.statsd.stats] ) for stat in stats: k, v = stat.split(':') if key == k: if kind is None: # key with no qualifiers is found return True s_value, s_kind = v.split('|') # if no kind match, look for other keys if kind != s_kind: continue if value: # special-case value|ms because statsd can turn # timing results into float of indeterminate # length, hence foiling string matching. if kind == 'ms': if float(value) >= float(s_value): return True elif value == s_value: return True # otherwise keep looking for other matches continue # this key matches return True time.sleep(0.1) raise Exception(f"Key {key} not found in reported stats") def assert_prometheus_stat(self, name, value, labels=None): sample_value = self._registry.get_sample_value(name, labels) self.assertEqual(sample_value, value) def test_list_projects(self): mock_uri = self.get_mock_url( service_type='identity', resource='projects', base_url_append='v3' ) self.register_uris( [ dict( method='GET', uri=mock_uri, status_code=200, json={'projects': []}, ) ] ) self.cloud.list_projects() self.assert_calls() self.assert_reported_stat( 'openstack.api.identity.GET.projects.200', value='1', kind='c' ) self.assert_prometheus_stat( 'openstack_http_requests_total', 1, dict( service_type='identity', endpoint=mock_uri, method='GET', status_code='200', ), ) def test_projects(self): mock_uri = self.get_mock_url( service_type='identity', resource='projects', base_url_append='v3' ) self.register_uris( [ dict( method='GET', uri=mock_uri, status_code=200, json={'projects': []}, ) ] ) list(self.cloud.identity.projects()) self.assert_calls() self.assert_reported_stat( 'openstack.api.identity.GET.projects.200', value='1', kind='c' ) self.assert_prometheus_stat( 'openstack_http_requests_total', 1, dict( service_type='identity', endpoint=mock_uri, method='GET', status_code='200', ), ) def test_servers(self): mock_uri = 'https://compute.example.com/v2.1/servers/detail' self.register_uris( [ self.get_nova_discovery_mock_dict(), dict( method='GET', uri=mock_uri, status_code=200, json={'servers': []}, ), ] ) list(self.cloud.compute.servers()) self.assert_calls() self.assert_reported_stat( 'openstack.api.compute.GET.servers_detail.200', value='1', kind='c' ) self.assert_reported_stat( 'openstack.api.compute.GET.servers_detail.200', value='5', kind='ms', ) self.assert_prometheus_stat( 'openstack_http_requests_total', 1, dict( service_type='compute', endpoint=mock_uri, method='GET', status_code='200', ), ) def test_servers_no_detail(self): mock_uri = 'https://compute.example.com/v2.1/servers' self.register_uris( [ dict( method='GET', uri=mock_uri, status_code=200, json={'servers': []}, ) ] ) self.cloud.compute.get('/servers') self.assert_calls() self.assert_reported_stat( 'openstack.api.compute.GET.servers.200', value='1', kind='c' ) self.assert_reported_stat( 'openstack.api.compute.GET.servers.200', value='5', kind='ms' ) self.assert_reported_stat( 'openstack.api.compute.GET.servers.attempted', value='1', kind='c' ) self.assert_prometheus_stat( 'openstack_http_requests_total', 1, dict( service_type='compute', endpoint=mock_uri, method='GET', status_code='200', ), ) def test_servers_error(self): mock_uri = 'https://compute.example.com/v2.1/servers' self.register_uris( [dict(method='GET', uri=mock_uri, status_code=500, json={})] ) self.cloud.compute.get('/servers') self.assert_calls() self.assert_reported_stat( 'openstack.api.compute.GET.servers.500', value='1', kind='c' ) self.assert_reported_stat( 'openstack.api.compute.GET.servers.500', value='5', kind='ms' ) self.assert_reported_stat( 'openstack.api.compute.GET.servers.attempted', value='1', kind='c' ) self.assert_prometheus_stat( 'openstack_http_requests_total', 1, dict( service_type='compute', endpoint=mock_uri, method='GET', status_code='500', ), ) def test_timeout(self): mock_uri = 'https://compute.example.com/v2.1/servers' self.register_uris( [dict(method='GET', uri=mock_uri, exc=rexceptions.ConnectTimeout)] ) try: self.cloud.compute.get('/servers') except exceptions.ConnectTimeout: pass self.assert_reported_stat( 'openstack.api.compute.GET.servers.failed', value='1', kind='c' ) self.assert_reported_stat( 'openstack.api.compute.GET.servers.attempted', value='1', kind='c' ) class TestNoStats(base.TestCase): def setUp(self): super().setUp() self.statsd = StatsdFixture() self.useFixture(self.statsd) def test_no_stats(self): mock_uri = self.get_mock_url( service_type='identity', resource='projects', base_url_append='v3' ) self.register_uris( [ dict( method='GET', uri=mock_uri, status_code=200, json={'projects': []}, ) ] ) self.cloud.identity._statsd_client = None list(self.cloud.identity.projects()) self.assert_calls() self.assertEqual([], self.statsd.stats)