Added template match check when applying template to metric path for templates without filters. Corrected tests that were wrongly expecting data from branch nodes. Updated finder to fail fast on invalid node requests. Updated finder to not query for paths without a template match

Author: Dan

influxgraph/classes/finder.py

@@ -44,7 +44,8 @@
 from ..utils import NullStatsd, calculate_interval, read_influxdb_values, \
      get_aggregation_func, gen_memcache_key, gen_memcache_pattern_key, \
      Query, get_retention_policy, _compile_aggregation_patterns, parse_series
-from ..templates import parse_influxdb_graphite_templates, apply_template
+from ..templates import parse_influxdb_graphite_templates, apply_template, \
+     TemplateMatchError
 from .reader import InfluxDBReader
 from .leaf import InfluxDBLeafNode
 try:
@@ -356,8 +357,11 @@ def _get_all_template_values(self, paths):
             for path in paths:
                 if _filter and not _filter.match(path):
                     continue
-                measurement, tags, field = apply_template(
-                    path.split('.'), template, default_tags, separator)
+                try:
+                    measurement, tags, field = apply_template(
+                        path.split('.'), template, default_tags, separator)
+                except TemplateMatchError:
+                    continue
                 if measurement not in _measurements:
                     _measurements.append(measurement)
                 for tag in tags:
@@ -380,6 +384,8 @@ def _get_all_template_values(self, paths):
 
     def _gen_query_values_from_templates(self, paths, retention):
         _query_data, path_measurements = self._get_all_template_values(paths)
+        if len(_query_data) == 0:
+            return
         query_data = deque()
         for (_measurements, _tags, _fields) in _query_data:
             measurements = ', '.join(
@@ -410,7 +416,10 @@ def _gen_influxdb_query(self, start_time, end_time, paths, interval):
         memcache_key = gen_memcache_key(start_time, end_time, aggregation_func,
                                         paths)
         queries = deque()
-        query_data, path_measurements = self._gen_query_values(paths, retention)
+        try:
+            query_data, path_measurements = self._gen_query_values(paths, retention)
+        except TypeError:
+            return
         for (measurement, tags, _fields) in query_data:
             if not _fields:
                 _fields = ['value']
@@ -434,6 +443,12 @@ def _gen_influxdb_query(self, start_time, end_time, paths, interval):
         query = ';'.join(queries)
         return query, memcache_key, path_measurements
 
+    def _make_empty_multi_fetch_result(self, time_info, paths):
+        data = {}
+        for key in paths:
+            data[key] = []
+        return time_info, data
+
     def fetch_multi(self, nodes, start_time, end_time):
         """Fetch datapoints for all series between start and end times
 
@@ -446,9 +461,15 @@ def fetch_multi(self, nodes, start_time, end_time):
         time_info = start_time, end_time, interval
         if not nodes:
             return time_info, {}
-        paths = [n.path for n in nodes]
-        query, memcache_key, path_measurements = self._gen_influxdb_query(
-            start_time, end_time, paths, interval)
+        paths = [n.path for n in nodes if n.is_leaf]
+        if not len(paths) > 0:
+            return self._make_empty_multi_fetch_result(
+                time_info, [n.path for n in nodes])
+        try:
+            query, memcache_key, path_measurements = self._gen_influxdb_query(
+                start_time, end_time, paths, interval)
+        except TypeError:
+            return self._make_empty_multi_fetch_result(time_info, paths)
         data = self.memcache.get(memcache_key) if self.memcache else None
         if data:
             logger.debug("Found cached data for key %s", memcache_key)

influxgraph/templates.py

@@ -49,6 +49,11 @@ class InvalidTemplateError(Exception):
     pass
 
 
+class TemplateMatchError(Exception):
+    """Raised on errors matching template with path"""
+    pass
+
+
 # Function as per Python official documentation
 def heapsort(iterable):
     h = []
@@ -118,10 +123,14 @@ def _template_sanity_check(template):
             template)
 
 def apply_template(metric_path_parts, template, default_tags, separator='.'):
-    """Apply template to metric path parts and return measurements, tags and field"""
+    """Apply template to metric path parts and return measurements, tags and field
+
+    :raises: mod:`TemplateMatchError` on error matching template"""
     measurement = []
     tags = {}
     field = ""
+    if len(metric_path_parts) < len(template.keys()):
+        raise TemplateMatchError()
     for i, tag in template.items():
         if i >= len(metric_path_parts):
             continue

tests/test_influxdb_integration.py

@@ -364,7 +364,8 @@ def test_multi_fetch_non_existant_series(self):
             database=self.db_name), path1, influxgraph.utils.NullStatsd())
         reader2 = influxgraph.InfluxdbReader(InfluxDBClient(
             database=self.db_name), path2, influxgraph.utils.NullStatsd())
-        nodes = [reader1, reader2]
+        nodes = [influxgraph.classes.leaf.InfluxDBLeafNode(path1, reader1),
+                 influxgraph.classes.leaf.InfluxDBLeafNode(path2, reader2)]
         time_info, data = self.finder.fetch_multi(nodes,
                                                   int(self.start_time.strftime("%s")),
                                                   int(self.end_time.strftime("%s")))

tests/test_influxdb_templates_integration.py

@@ -353,27 +353,27 @@ def test_template_multi_tag_no_field(self):
                      "*.interface.* host.measurement.device.metric",
                      ]
         mem_measurement = 'memory'
-        tags = {'host': 'my_host',
-                'metric': 'some_metric',
+        mem_tags = {'host': 'my_host',
+                'metric': 'mem_metric',
                 }
         fields = {'value': 1,
                   }
-        self.write_data([mem_measurement], tags, fields)
+        self.write_data([mem_measurement], mem_tags, fields)
         int_measurement = 'interface'
-        tags = {'host': 'my_host',
-                'device': 'dev',
-                'metric': 'some_metric',
-                }
+        int_tags = {'host': 'my_host',
+                    'device': 'dev',
+                    'metric': 'int_metric',
+                    }
         fields = {'value': 1,
                   }
-        self.write_data([int_measurement], tags, fields)
+        self.write_data([int_measurement], int_tags, fields)
         self.config['influxdb']['templates'] = templates
         self.finder = influxgraph.InfluxDBFinder(self.config)
         ##
         query = Query('%s.%s.*' % (
-            tags['host'], mem_measurement,))
+            mem_tags['host'], mem_measurement,))
         nodes = list(self.finder.find_nodes(query))
-        self.assertEqual(sorted([n.name for n in nodes]), [tags['metric']])
+        self.assertEqual(sorted([n.name for n in nodes]), [mem_tags['metric']])
         time_info, data = self.finder.fetch_multi(
             nodes, int(self.start_time.strftime("%s")),
             int(self.end_time.strftime("%s")))
@@ -383,29 +383,22 @@ def test_template_multi_tag_no_field(self):
                             self.num_datapoints, nodes[0].path, len(datapoints),))
         ##
         query = Query('%s.%s.*' % (
-            tags['host'], int_measurement,))
+            int_tags['host'], int_measurement,))
         nodes = list(self.finder.find_nodes(query))
-        self.assertEqual(sorted([n.name for n in nodes]), [tags['device']])
+        self.assertEqual(sorted([n.name for n in nodes]), [int_tags['device']])
         time_info, data = self.finder.fetch_multi(
             nodes, int(self.start_time.strftime("%s")),
             int(self.end_time.strftime("%s")))
         for metric in [n.path for n in nodes]:
-            datapoints = [v for v in data[metric] if v]
-            self.assertTrue(len(datapoints) == self.num_datapoints,
-                            msg="Expected %s datapoints for %s - got %s" % (
-                                self.num_datapoints, metric, len(datapoints),))
+            self.assertTrue(len(data[metric]) == 0)
         query = Query('%s.%s.*.*' % (
-            tags['host'], int_measurement,))
+            int_tags['host'], int_measurement,))
         nodes = list(self.finder.find_nodes(query))
-        self.assertEqual(sorted([n.name for n in nodes]), [tags['metric']])
+        self.assertEqual(sorted([n.name for n in nodes]), [int_tags['metric']])
         time_info, data = self.finder.fetch_multi(
             nodes, int(self.start_time.strftime("%s")),
             int(self.end_time.strftime("%s")))
-        for metric in [n.path for n in nodes]:
-            datapoints = [v for v in data[metric] if v]
-            self.assertTrue(len(datapoints) == self.num_datapoints,
-                            msg="Expected %s datapoints for %s - got %s" % (
-                                self.num_datapoints, metric, len(datapoints),))
+        self._test_data_in_nodes(nodes)
 
     def test_template_multiple_tags(self):
         self.client.drop_database(self.db_name)
@@ -504,18 +497,19 @@ def test_data_with_fields(self):
         measurements = ['cpu-0', 'cpu-1', 'cpu-2', 'cpu-3']
         fields = {'load': 1, 'idle': 1,
                   'usage': 1, 'user': 1,
-                  'user.io': 1, 'idle.io': 1,
-                  'load.io': 1, 'usage.io': 1,
-        }
+                  'io.usr': 1, 'io.swp': 1,
+                  'io.sys': 1,
+                  }
         tags = {'host': 'my_host',
                 'env': 'my_env',
                 }
         cpu_metrics = ['.'.join(['my_host', m, f])
                        for m in measurements
-                       for f in ['load', 'usage', 'user', 'idle']]
+                       for f in ['load', 'usage', 'user', 'idle',
+                                 'io']]
         io_metrics = ['.'.join(['my_host', m, f])
                        for m in measurements
-                       for f in ['user.io', 'idle.io', 'load.io', 'usage.io']]
+                       for f in ['io.usr', 'io.swp', 'io.sys']]
         self.client.drop_database(self.db_name)
         self.client.create_database(self.db_name)
         self.write_data(measurements, tags, fields)
@@ -530,13 +524,12 @@ def test_data_with_fields(self):
         nodes = list(self.finder.find_nodes(query))
         node_paths = sorted([n.path for n in nodes])
         expected = sorted(cpu_metrics)
-        self.assertEqual(node_paths, expected,
-                         msg="Got query %s result %s - wanted %s" % (
-                             query.pattern, node_paths, expected,))
+        self.assertEqual(node_paths, expected)
         time_info, data = self.finder.fetch_multi(nodes,
                                                   int(self.start_time.strftime("%s")),
                                                   int(self.end_time.strftime("%s")))
-        for metric in cpu_metrics:
+        cpu_leaf_nodes = [m for m in cpu_metrics if not m.endswith('io')]
+        for metric in cpu_leaf_nodes:
             self.assertTrue(metric in data,
                             msg="Did not get data for requested series %s - got data for %s" % (
                                 metric, data.keys(),))
@@ -547,11 +540,15 @@ def test_data_with_fields(self):
         time_info, _data = self.finder.fetch_multi(nodes,
                                                    int(self.start_time.strftime("%s")),
                                                    int(self.end_time.strftime("%s")))
-        for metric in cpu_metrics:
+        for metric in cpu_leaf_nodes:
             datapoints = [v for v in _data[metric] if v]
             self.assertTrue(len(datapoints) == self.num_datapoints,
                             msg="Expected %s datapoints for %s - got %s" % (
                                 self.num_datapoints, metric, len(datapoints),))
+        query = Query('%s.*.*.*' % (tags['host'],))
+        nodes = list(self.finder.find_nodes(query))
+        self.assertEqual(sorted([n.path for n in nodes]), sorted(io_metrics))
+        self._test_data_in_nodes(nodes)
 
     def test_multi_tag_values_multi_measurements(self):
         measurements = ['cpu-0', 'cpu-1', 'cpu-2', 'cpu-3']