Small data collection validation refinements

collectors/0/elasticsearch.py

@@ -182,7 +182,10 @@ def _collect_server(server, version, lock):
   tags = {"cluster": cluster_name, "node": node_name}
   #tags.update(nstats["attributes"])
 
-  is_master = nodeid == cluster_master_node(server)
+  if nodeid == cluster_master_node(server):
+      is_master = 1
+  else:
+      is_master = 0
   with lock:
     printmetric(rootmetric + ".is_master", ts, is_master, tags)
   if is_master:

collectors/0/tcollector.py

@@ -147,7 +147,7 @@ def collect_tcollect_stats(processes):
 
         # if <collector> is executed with "python <collector.py>",
         # ensure <collector.py> is used as name
-        if p.comm == "python":
+        if p.comm == "python" and p.cmdline is not None:
             comm = p.cmdline[1].split('/')[-1]
         else:
             comm = p.comm

tcollector.py

@@ -351,14 +351,20 @@ def process_line(self, col, line):
             col.lines_invalid += 1
             return
         metric, timestamp, value, tags = parsed.groups()
+        try:
+            # The regex above is fairly open, and would leave values like 'True' through
+            testvalue = float(value)
+        except:
+            LOG.warning('%s sent invalid value: %s', col.name, line)
+            col.lines_invalid += 1
+            return
 
         # If there are more than 11 digits we're dealing with a timestamp
         # with millisecond precision
         max_timestamp = MAX_REASONABLE_TIMESTAMP
         if len(timestamp) > 11:
             max_timestamp = MAX_REASONABLE_TIMESTAMP * 1000
-
-        timestamp = int(timestamp)
+        timestamp = float(timestamp)
 
         # De-dupe detection...  To reduce the number of points we send to the
         # TSD, we suppress sending values of metrics that don't change to

tests.py

@@ -31,6 +31,35 @@ def always_true():
     return True
 
 
+class ReaderThreadTests(unittest.TestCase):
+
+    def test_bad_float(self):
+        """Values that aren't ints/floats aren't sent to OpenTSDB.
+
+        This can happen if a specific collector is buggy.
+        """
+        thread = tcollector.ReaderThread(1, 10, True)
+        collector = tcollector.Collector("c", 1, "c")
+        for line in ["xxx", "mymetric 123 True a=b"]:
+            thread.process_line(collector, line)
+        self.assertEqual(thread.readerq.qsize(), 0)
+        self.assertEqual(collector.lines_received, 2)
+        self.assertEqual(collector.lines_invalid, 2)
+
+    def test_ok_lines(self):
+        """Good lines are passed on to OpenTSDB."""
+        thread = tcollector.ReaderThread(1, 10, True)
+        collector = tcollector.Collector("c", 1, "c")
+        for line in ["mymetric 123.24 12 a=b",
+                     "mymetric 124 12.7 a=b",
+                     "mymetric 125 12.7"]:
+            thread.process_line(collector, line)
+            self.assertEqual(thread.readerq.qsize(), 1, line)
+            self.assertEqual(thread.readerq.get(), line)
+        self.assertEqual(collector.lines_received, 3)
+        self.assertEqual(collector.lines_invalid, 0)
+
+
 class CollectorsTests(unittest.TestCase):
 
     def test_collectorsAccessRights(self):