Merge pull request #2757 from ross/consul-net-latency

Consul net latency

Author: truthbk

checks.d/consul.py

@@ -6,6 +6,7 @@
 from collections import defaultdict
 from datetime import datetime, timedelta
 from itertools import islice
+from math import ceil, floor, sqrt
 from urlparse import urljoin
 
 # project
@@ -15,6 +16,29 @@
 import requests
 
 
+# More information in https://www.consul.io/docs/internals/coordinates.html,
+# code is based on the snippet there.
+def distance(a, b):
+    a = a['Coord']
+    b = b['Coord']
+    total = 0
+    b_vec = b['Vec']
+    for i, a_p in enumerate(a['Vec']):
+        diff = a_p - b_vec[i]
+        total += diff * diff
+    rtt = sqrt(total) + a['Height'] + b['Height']
+
+    adjusted = rtt + a['Adjustment'] + b['Adjustment']
+    if adjusted > 0.0:
+        rtt = adjusted
+
+    return rtt * 1000.0
+
+
+def ceili(v):
+    return int(ceil(v))
+
+
 class ConsulCheck(AgentCheck):
     CONSUL_CHECK = 'consul.up'
     HEALTH_CHECK = 'consul.check'
@@ -102,7 +126,18 @@ def _is_instance_leader(self, instance):
             return False
 
     def _check_for_leader_change(self, instance):
-        agent_dc = self._get_agent_datacenter(instance)
+        perform_new_leader_checks = instance.get('new_leader_checks',
+                                                 self.init_config.get('new_leader_checks', False))
+        perform_self_leader_check = instance.get('self_leader_check',
+                                                 self.init_config.get('self_leader_check', False))
+
+        if perform_new_leader_checks and perform_self_leader_check:
+            self.log.warn('Both perform_self_leader_check and perform_new_leader_checks are set, '
+                          'ignoring perform_new_leader_checks')
+        elif not perform_new_leader_checks and not perform_self_leader_check:
+            # Nothing to do here
+            return
+
         leader = self._get_cluster_leader(instance)
 
         if not leader:
@@ -118,24 +153,30 @@ def _check_for_leader_change(self, instance):
             self._last_known_leader = leader
             return
 
+        agent = self._get_agent_url(instance)
+        agent_dc = self._get_agent_datacenter(instance)
+
         if leader != self._last_known_leader:
-            self.log.info(('Leader change from {0} to {1}. Sending new leader event').format(
-                self._last_known_leader, leader))
-
-            self.event({
-                "timestamp": int(datetime.now().strftime("%s")),
-                "event_type": "consul.new_leader",
-                "source_type_name": self.SOURCE_TYPE_NAME,
-                "msg_title": "New Consul Leader Elected in consul_datacenter:{0}".format(agent_dc),
-                "aggregation_key": "consul.new_leader",
-                "msg_text": "The Node at {0} is the new leader of the consul datacenter {1}".format(
-                    leader,
-                    agent_dc
-                ),
-                "tags": ["prev_consul_leader:{0}".format(self._last_known_leader),
-                         "curr_consul_leader:{0}".format(leader),
-                         "consul_datacenter:{0}".format(agent_dc)]
-            })
+            # There was a leadership change
+            if perform_new_leader_checks or (perform_self_leader_check and agent == leader):
+                # We either emit all leadership changes or emit when we become the leader and that just happened
+                self.log.info(('Leader change from {0} to {1}. Sending new leader event').format(
+                    self._last_known_leader, leader))
+
+                self.event({
+                    "timestamp": int(datetime.now().strftime("%s")),
+                    "event_type": "consul.new_leader",
+                    "source_type_name": self.SOURCE_TYPE_NAME,
+                    "msg_title": "New Consul Leader Elected in consul_datacenter:{0}".format(agent_dc),
+                    "aggregation_key": "consul.new_leader",
+                    "msg_text": "The Node at {0} is the new leader of the consul datacenter {1}".format(
+                        leader,
+                        agent_dc
+                    ),
+                    "tags": ["prev_consul_leader:{0}".format(self._last_known_leader),
+                             "curr_consul_leader:{0}".format(leader),
+                             "consul_datacenter:{0}".format(agent_dc)]
+                })
 
         self._last_known_leader = leader
 
@@ -167,10 +208,7 @@ def _cull_services_list(self, services, service_whitelist):
         return services
 
     def check(self, instance):
-        perform_new_leader_checks = instance.get('new_leader_checks',
-                                                 self.init_config.get('new_leader_checks', False))
-        if perform_new_leader_checks:
-            self._check_for_leader_change(instance)
+        self._check_for_leader_change(instance)
 
         peers = self.get_peers_in_cluster(instance)
         main_tags = []
@@ -190,6 +228,8 @@ def check(self, instance):
         service_check_tags = ['consul_url:{0}'.format(instance.get('url'))]
         perform_catalog_checks = instance.get('catalog_checks',
                                               self.init_config.get('catalog_checks'))
+        perform_network_latency_checks = instance.get('network_latency_checks',
+                                                      self.init_config.get('network_latency_checks'))
 
         try:
             # Make service checks from health checks for all services in catalog
@@ -323,3 +363,69 @@ def check(self, instance):
                         status_value,
                         tags=main_tags+node_tags
                     )
+
+        if perform_network_latency_checks:
+            self.check_network_latency(instance, agent_dc, main_tags)
+
+    def _get_coord_datacenters(self, instance):
+        return self.consul_request(instance, '/v1/coordinate/datacenters')
+
+    def _get_coord_nodes(self, instance):
+        return self.consul_request(instance, 'v1/coordinate/nodes')
+
+    def check_network_latency(self, instance, agent_dc, main_tags):
+
+        datacenters = self._get_coord_datacenters(instance)
+        for datacenter in datacenters:
+            name = datacenter['Datacenter']
+            if name == agent_dc:
+                # This is us, time to collect inter-datacenter data
+                for other in datacenters:
+                    other_name = other['Datacenter']
+                    if name == other_name:
+                        # Ignore ourself
+                        continue
+                    latencies = []
+                    for node_a in datacenter['Coordinates']:
+                        for node_b in other['Coordinates']:
+                            latencies.append(distance(node_a, node_b))
+                    latencies.sort()
+                    tags = main_tags + ['source_datacenter:{}'.format(name),
+                                        'dest_datacenter:{}'.format(other_name)]
+                    n = len(latencies)
+                    half_n = int(floor(n / 2))
+                    if n % 2:
+                        median = latencies[half_n]
+                    else:
+                        median = (latencies[half_n - 1] + latencies[half_n]) / 2
+                    self.gauge('consul.net.dc.latency.min', latencies[0], hostname='', tags=tags)
+                    self.gauge('consul.net.dc.latency.median', median, hostname='', tags=tags)
+                    self.gauge('consul.net.dc.latency.max', latencies[-1], hostname='', tags=tags)
+                # We've found ourself, we can move on
+                break
+
+        # Intra-datacenter
+        nodes = self._get_coord_nodes(instance)
+        for node in nodes:
+            node_name = node['Node']
+            latencies = []
+            for other in nodes:
+                other_name = other['Node']
+                if node_name == other_name:
+                    continue
+                latencies.append(distance(node, other))
+            latencies.sort()
+            n = len(latencies)
+            half_n = int(floor(n / 2))
+            if n % 2:
+                median = latencies[half_n]
+            else:
+                median = (latencies[half_n - 1] + latencies[half_n]) / 2
+            self.gauge('consul.net.node.latency.min', latencies[0], hostname=node_name, tags=main_tags)
+            self.gauge('consul.net.node.latency.p25', latencies[ceili(n * 0.25) - 1], hostname=node_name, tags=main_tags)
+            self.gauge('consul.net.node.latency.median', median, hostname=node_name, tags=main_tags)
+            self.gauge('consul.net.node.latency.p75', latencies[ceili(n * 0.75) - 1], hostname=node_name, tags=main_tags)
+            self.gauge('consul.net.node.latency.p90', latencies[ceili(n * 0.90) - 1], hostname=node_name, tags=main_tags)
+            self.gauge('consul.net.node.latency.p95', latencies[ceili(n * 0.95) - 1], hostname=node_name, tags=main_tags)
+            self.gauge('consul.net.node.latency.p99', latencies[ceili(n * 0.99) - 1], hostname=node_name, tags=main_tags)
+            self.gauge('consul.net.node.latency.max', latencies[-1], hostname=node_name, tags=main_tags)

conf.d/consul.yaml.example

@@ -23,11 +23,25 @@ instances:
       # Whether to perform checks against the Consul service Catalog
       catalog_checks: yes
 
+      # Whether to enable self leader checks. Each agent with this enabled will
+      # watch for itself to become the leader and will emit an event when that
+      # happens. It is safe/expected to enable this on all nodes in a consul
+      # cluster since only the new leader will emit the (single) event. This
+      # flag takes precedence over new_leader_checks.
+      self_leader_check: yes
+
       # Whether to enable new leader checks from this agent
       # Note: if this is set on multiple agents in the same cluster
-      # you will receive one event per leader change per agent
+      # you will receive one event per leader change per agent. See
+      # self_leader_check for a more robust option.
       new_leader_checks: yes
 
+      # Whether to enable network latency metrics collection. When enabled
+      # consul network coordinates will be retrieved and latency calculated for
+      # each node and between data centers.
+      # See https://www.consul.io/docs/internals/coordinates.html
+      network_latency_checks: yes
+
       # Services to restrict catalog querying to
       # The default settings query up to 50 services. So if you have more than
       # this many in your Consul service catalog, you will want to fill in the