Start of obtaining worker stats.

The worker now gets the bytes-in/bytes-out for the specified
protocol (http or tcp). For now, these values are only logged,
and then only in debug mode. Added a test for setting these
values in the new LBStatistics class.

Updated README for the 'socat' requirement and to mention that
the PID file directory must be writable by the user.

Change-Id: I79f218747255cba84b25c4a69d0b210c9d1dfee5

README

@@ -31,6 +31,12 @@ Now you may install the Libra toolset:
 
     $ sudo python setup.py install
 
+The worker also needs some packages installed in order to be used with
+HAProxy. The commands below will install them on Ubuntu:
+
+    $ sudo apt-get install haproxy
+    $ sudo apt-get install socat
+
 
 Edit /etc/sudoers
 -----------------
@@ -40,7 +46,7 @@ prompted for a password. It is suggested that you run the worker as
 the `haproxy` user and `haproxy` group on Ubuntu systems. Then add the
 following line to /etc/sudoers:
 
-    %haproxy ALL = NOPASSWD: /usr/sbin/service, /bin/cp, /bin/mv, /bin/rm
+    %haproxy ALL = NOPASSWD: /usr/sbin/service, /bin/cp, /bin/mv, /bin/rm, /usr/bin/socat
 
 The above lets everyone in the `haproxy` group run those commands
 as root without being prompted for a password.
@@ -65,6 +71,10 @@ Basic commands:
     # Start up with debugging output in non-daemon mode
     $ libra_worker --debug --nodaemon
 
+NOTE: When running the worker in daemon mode, you must make sure that the
+directory where the PID file will be (-p/--pid option) is writable by the
+user/group specified with the --user and --group options.
+
 You can verify that the worker is running by using the sample Gearman
 client in the bin/ directory:
 

libra/common/lbstats.py

@@ -15,4 +15,28 @@
 
 class LBStatistics(object):
     """ Load balancer statistics class. """
-    pass
+
+    def __init__(self):
+        self.stats = {}
+        self.bytes_out = 0
+        self.bytes_in = 0
+
+    @property
+    def bytes_out(self):
+        return self.stats['bytes_out']
+
+    @bytes_out.setter
+    def bytes_out(self, value):
+        if not isinstance(value, int):
+            raise TypeError("Must be an integer.")
+        self.stats['bytes_out'] = value
+
+    @property
+    def bytes_in(self):
+        return self.stats['bytes_in']
+
+    @bytes_in.setter
+    def bytes_in(self, value):
+        if not isinstance(value, int):
+            raise TypeError("Must be an integer.")
+        self.stats['bytes_in'] = value

libra/worker/drivers/base.py

@@ -72,6 +72,6 @@ class LoadBalancerDriver(object):
         """ Delete a load balancer. """
         raise NotImplementedError()
 
-    def get_stats(self):
-        """ Get load balancer statistics. """
+    def get_stats(self, protocol):
+        """ Get load balancer statistics for specified protocol. """
         raise NotImplementedError()

libra/worker/drivers/haproxy/driver.py

@@ -153,5 +153,5 @@ class HAProxyDriver(LoadBalancerDriver):
         self.ossvc.service_stop()
         self.ossvc.remove_configs()
 
-    def get_stats(self):
-        return self.ossvc.get_stats()
+    def get_stats(self, protocol):
+        return self.ossvc.get_stats(protocol)

libra/worker/drivers/haproxy/ubuntu_services.py

@@ -12,6 +12,7 @@
 # License for the specific language governing permissions and limitations
 # under the License.
 
+import csv
 import os
 import subprocess
 
@@ -102,6 +103,45 @@ class UbuntuServices(ServicesBase):
             raise Exception("Failed to delete HAProxy config files: %s" %
                             e.output.rstrip('\n'))
 
-    def get_stats(self):
+    def get_stats(self, protocol):
+        """
+        Query HAProxy socket for stats on the given protocol.
+
+        protocol
+            One of the supported protocol names (http or tcp).
+
+        This function will query the HAProxy statistics socket and pull out
+        the values that it needs for the given protocol (which equates to one
+        load balancer). The values are stored in a LBStatistics object that
+        will be returned to the caller.
+
+        The output of the socket query is in CSV format and defined here:
+
+        http://cbonte.github.com/haproxy-dconv/configuration-1.4.html#9
+        """
+
         stats = LBStatistics()
+
+        cmd = 'echo "show stat" | ' \
+              'sudo /usr/bin/socat stdio /var/run/haproxy-stats.socket'
+        try:
+            csv_output = subprocess.check_output(cmd, shell=True)
+        except subprocess.CalledProcessError as e:
+            raise Exception("Failed to get statistics: %s" %
+                            e.output.rstrip('\n'))
+
+        # Remove leading '# ' from string and trailing newlines
+        csv_output = csv_output[2:].rstrip()
+        # Turn string into a list, removing last two empty lines
+        csv_lines = csv_output.split('\n')
+
+        proxy_name = "%s-in" % protocol.lower()
+        service_name = "FRONTEND"
+
+        reader = csv.DictReader(csv_lines)
+        for row in reader:
+            if row['pxname'] == proxy_name and row['svname'] == service_name:
+                stats.bytes_out = row['bout']
+                break
+
         return stats

libra/worker/stats_client.py

@@ -21,14 +21,18 @@ def stats_manager(logger, driver, stats_poll):
 
     while True:
         try:
-            stats = driver.get_stats()
+            http_stats = driver.get_stats('http')
+            tcp_stats = driver.get_stats('tcp')
         except NotImplementedError:
             logger.warn(
                 "[stats] Driver does not implement statisics gathering."
             )
             break
 
-        logger.debug("[stats] Statistics: %s" % stats)
+        logger.debug("[stats] HTTP bytes in/out: (%d, %d)" %
+                     (http_stats.bytes_in, http_stats.bytes_out))
+        logger.debug("[stats] TCP bytes in/out: (%d, %d)" %
+                     (tcp_stats.bytes_in, tcp_stats.bytes_out))
         eventlet.sleep(stats_poll)
 
     logger.info("[stats] Statistics gathering process terminated.")

tests/test_lbstats.py

@@ -0,0 +1,26 @@
+import unittest
+from libra.common.lbstats import LBStatistics
+
+
+class TestLBStatistics(unittest.TestCase):
+    def setUp(self):
+        self.stats = LBStatistics()
+
+    def tearDown(self):
+        pass
+
+    def testInitValues(self):
+        self.assertEquals(self.stats.bytes_out, 0)
+        self.assertEquals(self.stats.bytes_in, 0)
+
+    def testSetBytesIn(self):
+        self.stats.bytes_in = 99
+        self.assertEquals(self.stats.bytes_in, 99)
+        with self.assertRaises(TypeError):
+            self.stats.bytes_in = "NaN"
+
+    def testSetBytesOut(self):
+        self.stats.bytes_out = 100
+        self.assertEquals(self.stats.bytes_out, 100)
+        with self.assertRaises(TypeError):
+            self.stats.bytes_out = "NaN"