Exercise statsd in tests and fix

We weren't doing anything with statsd in tests.  Port over the
fake statsd from Zuul and use it to verify that we exit some
stats.

Fix parts of the stats emission that were broken.

Change-Id: I027e67b928bd28372bef8ab147c7ed5841009caf

nodepool/nodepool.py

@@ -103,8 +103,12 @@ class StatsReporter(object):
                         (provider_name, node_az, subkey))
 
         if requestor:
-           keys.append('nodepool.launch.requestor.%s.%s' %
-                       (requestor, subkey))
+            # Replace '.' which is a graphite hierarchy, and ':' which is
+            # a statsd delimeter.
+            requestor = requestor.replace('.', '_')
+            requestor = requestor.replace(':', '_')
+            keys.append('nodepool.launch.requestor.%s.%s' %
+                        (requestor, subkey))
 
         for key in keys:
             self._statsd.timing(key, dt)
@@ -156,7 +160,8 @@ class StatsReporter(object):
 
         #nodepool.provider.PROVIDER.max_servers
         key = 'nodepool.provider.%s.max_servers' % provider.name
-        self._statsd.gauge(key, provider.max_servers)
+        max_servers = sum([p.max_servers for p in provider.pools.values()])
+        self._statsd.gauge(key, max_servers)
 
 
 class InstanceDeleter(threading.Thread, StatsReporter):

nodepool/tests/__init__.py

@@ -19,7 +19,9 @@ import glob
 import logging
 import os
 import random
+import select
 import string
+import socket
 import subprocess
 import threading
 import tempfile
@@ -97,6 +99,39 @@ class ChrootedKazooFixture(fixtures.Fixture):
         _tmp_client.close()
 
 
+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()
+
+    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, '1\n')
+        self.thread.join()
+
+
 class BaseTestCase(testtools.TestCase):
     def setUp(self):
         super(BaseTestCase, self).setUp()
@@ -138,6 +173,14 @@ class BaseTestCase(testtools.TestCase):
             self.subprocesses.append(p)
             return p
 
+        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
+        os.environ['STATSD_HOST'] = '127.0.0.1'
+        os.environ['STATSD_PORT'] = str(self.statsd.port)
+
         self.useFixture(fixtures.MonkeyPatch('subprocess.Popen',
                                              LoggingPopenFactory))
         self.setUpFakes()
@@ -198,6 +241,24 @@ class BaseTestCase(testtools.TestCase):
                 return
             time.sleep(0.1)
 
+    def assertReportedStat(self, key, value=None, kind=None):
+        start = time.time()
+        while time.time() < (start + 5):
+            for stat in self.statsd.stats:
+                k, v = stat.split(':')
+                if key == k:
+                    if value is None and kind is None:
+                        return
+                    elif value:
+                        if value == v:
+                            return
+                    elif kind:
+                        if v.endswith('|' + kind):
+                            return
+            time.sleep(0.1)
+
+        raise Exception("Key %s not found in reported stats" % key)
+
 
 class BuilderFixture(fixtures.Fixture):
     def __init__(self, configfile, cleanup_interval):

nodepool/tests/test_nodepool.py

@@ -69,6 +69,8 @@ class TestNodepool(tests.DBTestCase):
         )
         self.zk.deleteNodeRequest(req)
         self.waitForNodeRequestLockDeletion(req.id)
+        self.assertReportedStat('nodepool.nodes.ready', '1|g')
+        self.assertReportedStat('nodepool.nodes.building', '0|g')
 
     def test_node_assignment_at_quota(self):
         '''