[haproxy] unix socket stats urls

checks.d/haproxy.py

@@ -6,7 +6,12 @@
 from collections import defaultdict
 import copy
 import re
+import socket
 import time
+try:
+    import urlparse
+except LoadError:
+    import urllib.parse as urlparse
 
 # 3rd party
 import requests
@@ -97,8 +102,20 @@ def __init__(self, name, init_config, agentConfig, instances=None):
 
     def check(self, instance):
         url = instance.get('url')
-        username = instance.get('username')
-        password = instance.get('password')
+        self.log.debug('Processing HAProxy data for %s' % url)
+
+        parsed_url = urlparse.urlparse(url)
+
+        if parsed_url.scheme == 'unix':
+            data = self._fetch_socket_data(parsed_url.path)
+
+        else:
+            username = instance.get('username')
+            password = instance.get('password')
+            verify = not _is_affirmative(instance.get('disable_ssl_validation', False))
+
+            data = self._fetch_url_data(url, username, password, verify)
+
         collect_aggregates_only = _is_affirmative(
             instance.get('collect_aggregates_only', True)
         )
@@ -127,12 +144,6 @@ def check(self, instance):
 
         custom_tags = instance.get('tags', [])
 
-        verify = not _is_affirmative(instance.get('disable_ssl_validation', False))
-
-        self.log.debug('Processing HAProxy data for %s' % url)
-
-        data = self._fetch_data(url, username, password, verify)
-
         process_events = instance.get('status_check', self.init_config.get('status_check', False))
 
         self._process_data(
@@ -147,19 +158,38 @@ def check(self, instance):
             custom_tags=custom_tags,
         )
 
-    def _fetch_data(self, url, username, password, verify):
-        ''' Hit a given URL and return the parsed json '''
+    def _fetch_url_data(self, url, username, password, verify):
+        ''' Hit a given http url and return the stats lines '''
         # Try to fetch data from the stats URL
 
         auth = (username, password)
         url = "%s%s" % (url, STATS_URL)
 
-        self.log.debug("HAProxy Fetching haproxy search data from: %s" % url)
+        self.log.debug("Fetching haproxy stats from url: %s" % url)
+
+        response = requests.get(url, auth=auth, headers=headers(self.agentConfig), verify=verify, timeout=self.default_integration_http_timeout)
+        response.raise_for_status()
+
+        return response.content.splitlines()
+
+    def _fetch_socket_data(self, socket_path):
+        ''' Hit a given stats socket and return the stats lines '''
+
+        self.log.debug("Fetching haproxy stats from socket: %s" % socket_path)
+
+        sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+        sock.connect(socket_path)
+        sock.send("show stat\r\n")
+
+        response = ""
+        output = sock.recv(8192)
+        while output:
+            response += output.decode("ASCII")
+            output = sock.recv(8192)
 
-        r = requests.get(url, auth=auth, headers=headers(self.agentConfig), verify=verify, timeout=self.default_integration_http_timeout)
-        r.raise_for_status()
+        sock.close()
 
-        return r.content.splitlines()
+        return response.splitlines()
 
     def _process_data(self, data, collect_aggregates_only, process_events, url=None,
                       collect_status_metrics=False, collect_status_metrics_by_host=False,
@@ -428,8 +458,11 @@ def _process_metrics(self, data, url, services_incl_filter=None,
         hostname = data['svname']
         service_name = data['pxname']
         back_or_front = data['back_or_front']
-        tags = ["type:%s" % back_or_front, "instance_url:%s" % url]
-        tags.append("service:%s" % service_name)
+        tags = [
+            "type:%s" % back_or_front,
+            "instance_url:%s" % url,
+            "service:%s" % service_name,
+        ]
         tags.extend(custom_tags)
 
         if self._is_service_excl_filtered(service_name, services_incl_filter,

ci/haproxy.rb

@@ -44,8 +44,13 @@ def haproxy_rootdir
 
     task before_script: ['ci:common:before_script'] do
       %w(haproxy haproxy-open).each do |name|
+        # Older haproxy doesn't support ENV interpolation
+        config = File.read("#{ENV['TRAVIS_BUILD_DIR']}/ci/resources/haproxy/#{name}.cfg")
+        config.gsub!('$VOLATILE_DIR', ENV['VOLATILE_DIR'])
+        File.write("#{ENV['VOLATILE_DIR']}/#{name}.cfg", config)
+
         pid = spawn("#{haproxy_rootdir}/haproxy", '-d', '-f',
-                    "#{ENV['TRAVIS_BUILD_DIR']}/ci/resources/haproxy/#{name}.cfg",
+                    "#{ENV['VOLATILE_DIR']}/#{name}.cfg",
                     out: '/dev/null')
         Process.detach(pid)
         sh %(echo #{pid} > $VOLATILE_DIR/#{name}.pid)

ci/resources/haproxy/haproxy.cfg

@@ -2,6 +2,7 @@
 global
     log 127.0.0.1 local0
     maxconn 4096
+    stats socket $VOLATILE_DIR/datadog-haproxy-stats.sock
 
 defaults
     log     global

conf.d/haproxy.yaml.example

@@ -5,6 +5,10 @@ instances:
     # username: username
     # password: password
     #
+    # or, with a unix stats or admin socket:
+    #
+    # url: unix:///var/run/haproxy.sock
+    #
     # The (optional) `status_check` paramater will instruct the check to
     # send events on status changes in the backend. This is DEPRECATED in
     # favor creation a monitor on the service check status and will be

tests/checks/integration/test_haproxy.py

@@ -96,6 +96,14 @@ def __init__(self, *args, **kwargs):
                 'collect_aggregates_only': False,
             }]
         }
+        self.unixsocket_path = os.path.join(os.environ['VOLATILE_DIR'], 'datadog-haproxy-stats.sock')
+        self.unixsocket_url = 'unix://{0}'.format(self.unixsocket_path)
+        self.config_unixsocket = {
+            'instances': [{
+                'url': self.unixsocket_url,
+                'collect_aggregates_only': False,
+            }]
+        }
 
     def _test_frontend_metrics(self, shared_tag):
         frontend_tags = shared_tag + ['type:FRONTEND', 'service:public']
@@ -205,3 +213,14 @@ def test_open_config(self):
         self.assertEquals(self.service_checks[0]['host_name'], '')
 
         self.coverage_report()
+
+    def test_unixsocket_config(self):
+        self.run_check_twice(self.config_unixsocket)
+
+        shared_tag = ['instance_url:{0}'.format(self.unixsocket_url)]
+
+        self._test_frontend_metrics(shared_tag)
+        self._test_backend_metrics(shared_tag)
+        self._test_service_checks()
+
+        self.coverage_report()