Merge lp:~lucio.torre/txstatsd/add-long-count into lp:~txstatsd-dev/txstatsd/distinct-plugin

[Lucio Torre]

now 60 day counts are pushed to graphite too.
the value only changes on startup and when the buckets are saved (nightly)

[Sidnei da Silva (sidnei)]

+1, renaming long_count to longterm_distinct_days.

review: Approve

[lp:~lucio.torre/txstatsd/add-long-count]

14. By Lucio Torre on 2012-06-27

renamed longterm_distinct_[count|days]

=== modified file 'distinctdb/distinctmetric.py'
--- distinctdb/distinctmetric.py	2012-04-26 18:16:25 +0000
+++ distinctdb/distinctmetric.py	2012-06-27 20:06:19 +0000
@@ -52,9 +52,9 @@
         resource.Resource.__init__(self)
         self.reporter = reporter
         self.putChild("top",
-            JSONMethodResource(self.reporter, "get_distinct_top_value"))
+            JSONMethodResource(self.reporter, "_get_distinct_top_value"))
         self.putChild("count",
-            JSONMethodResource(self.reporter, "get_distinct_count"))
+            JSONMethodResource(self.reporter, "_get_distinct_count"))
 
 
 class DistinctMetricReporter(object):
@@ -67,7 +67,8 @@
     periods = [5 * ONE_MINUTE, ONE_HOUR, ONE_DAY]
 
     def __init__(self, name, wall_time_func=time.time, prefix="",
-            bucket_size=ONE_DAY, dsn=None, redis_host=None, redis_port=None):
+            bucket_size=ONE_DAY, dsn=None, redis_host=None, redis_port=None,
+            longterm_distinct_days=60):
         """Construct a metric we expect to be periodically updated.
 
         @param name: Indicates what is being instrumented.
@@ -90,10 +91,15 @@
         self.build_bucket()
         self.redis_flush_lock = threading.Lock()
         self.redis_count = {}
+        self.longterm_distinct_count = None
+        self.longterm_distinct_days = longterm_distinct_days
 
         if redis_host != None:
             self.redis = redis.client.Redis(host=redis_host, port=redis_port)
 
+        if self.dsn is not None:
+            reactor.callInThread(self._update_longterm_distinct)
+
     def build_bucket(self, timestamp=None):
         self.max = 0
         self.bucket = {}
@@ -152,30 +158,6 @@
     def count_1day(self):
         return self.count(ONE_DAY)
 
-    def _save_bucket(self, bucket, bucket_no):
-        path = self.prefix + self.name
-        c = psycopg2.connect(self.dsn)
-        cr = c.cursor()
-        if self.metric_id is None:
-
-            cr.execute("SELECT * FROM paths WHERE path = %s", (path,))
-            row = cr.fetchone()
-            if row is None:
-                cr.execute("INSERT INTO paths (path) VALUES (%s) "
-                            "RETURNING (id)", (path,))
-                row = cr.fetchone()
-                cr.execute("commit")
-
-            self.metric_id = row[0]
-
-        for i, (k, v) in enumerate(bucket.iteritems()):
-            cr.execute("INSERT INTO points (path_id, bucket, value, count) "
-                "VALUES (%s, %s, %s, %s)", (self.metric_id, bucket_no,
-                                            k, v))
-            if i % 1000 == 0:
-                cr.execute("commit")
-        cr.execute("commit")
-
     def save_bucket(self, bucket, bucket_no):
         if self.dsn is not None:
             reactor.callInThread(self._save_bucket, bucket, bucket_no)
@@ -196,11 +178,46 @@
                  ".count_1hour": self.count_1hour(),
                  ".count_1day": self.count_1day(),
                  }
+
+        if self.longterm_distinct_count != None:
+            items[".count_%dday" % (self.longterm_distinct_days,)] = \
+                self.longterm_distinct_count
+
         for item, value in items.iteritems():
             metrics.append((self.prefix + self.name + item, value, timestamp))
         return metrics
 
-    def get_distinct_count(self, since, until):
+    def _save_bucket(self, bucket, bucket_no):
+        path = self.prefix + self.name
+        c = psycopg2.connect(self.dsn)
+        cr = c.cursor()
+        if self.metric_id is None:
+
+            cr.execute("SELECT * FROM paths WHERE path = %s", (path,))
+            row = cr.fetchone()
+            if row is None:
+                cr.execute("INSERT INTO paths (path) VALUES (%s) "
+                            "RETURNING (id)", (path,))
+                row = cr.fetchone()
+                cr.execute("commit")
+
+            self.metric_id = row[0]
+
+        for i, (k, v) in enumerate(bucket.iteritems()):
+            cr.execute("INSERT INTO points (path_id, bucket, value, count) "
+                "VALUES (%s, %s, %s, %s)", (self.metric_id, bucket_no,
+                                            k, v))
+            if i % 1000 == 0:
+                cr.execute("commit")
+        cr.execute("commit")
+        self._update_longterm_distinct()
+
+    def _update_longterm_distinct(self):
+        now = self.wall_time_func()
+        self.longterm_distinct_count = self._get_distinct_count(
+            now - self.longterm_distinct_days * ONE_DAY, now)
+
+    def _get_distinct_count(self, since, until):
         """Get a distinct count for a path between certain ranges."""
         since_bucket = self.get_bucket_no(since)
         until_bucket = self.get_bucket_no(until)
@@ -214,7 +231,7 @@
         row = cr.fetchone()
         return row[0]
 
-    def get_distinct_top_value(self, since, until, how_many=20):
+    def _get_distinct_top_value(self, since, until, how_many=20):
         """Get the top distinct values for a path between certain ranges."""
         since_bucket = self.get_bucket_no(since)
         until_bucket = self.get_bucket_no(until)

=== modified file 'distinctdb/tests/test_distinct.py'
--- distinctdb/tests/test_distinct.py	2012-04-26 18:16:25 +0000
+++ distinctdb/tests/test_distinct.py	2012-06-27 20:06:19 +0000
@@ -44,11 +44,11 @@
         self.called.append(("get_foo", ()))
         return "foo"
 
-    def get_distinct_count(self, since, until):
+    def _get_distinct_count(self, since, until):
         self.called.append(("get_distinct_count", (since, until)))
         return 42
 
-    def get_distinct_top_value(self, since, until, how_many=20):
+    def _get_distinct_top_value(self, since, until, how_many=20):
         self.called.append(("get_distinct_top_value", (since, until, how_many)))
         return [("one", 1), ("two", 1)]
 
@@ -119,13 +119,14 @@
         o = TestOptions()
         config_file = ConfigParser.RawConfigParser()
         config_file.readfp(StringIO("[statsd]\n\n[plugin_distinctdb]\n"
-            "dsn = dbdsn\nbucket_size = 100"))
+            "dsn = dbdsn\nbucket_size = 100\nlongterm_distinct_days = 30"))
         o.configure(config_file)
         dmf = distinctdbplugin.DistinctMetricFactory()
         dmf.configure(o)
         dmr = dmf.build_metric("foo", "bar", time.time)
         self.assertEquals(dmr.bucket_size, 100)
         self.assertEquals(dmr.dsn, "dbdsn")
+        self.assertEquals(dmr.longterm_distinct_days, 30)
 
 
 class TestJSONMethodResource(TestCase):
@@ -231,7 +232,7 @@
 
 class ErrorDummyReporter(object):
 
-    def get_distinct_count(self, *args, **kweargs):
+    def _get_distinct_count(self, *args, **kweargs):
         raise Exception("die!")
 
 
@@ -376,26 +377,26 @@
         # Make sure we're using different buckets
         self.assertNotEqual(b1, b2)
         self._create_test_data_points("one", b1)
-        count = dmr.get_distinct_count(t1, t2)
+        count = dmr._get_distinct_count(t1, t2)
         self.assertEqual(count, 1)
         self._create_test_data_points("two", b1)
-        count = dmr.get_distinct_count(t1, t2)
+        count = dmr._get_distinct_count(t1, t2)
         self.assertEqual(count, 2)
         # Add it again, make sure the distinct is filtering it out
         self._create_test_data_points("two", b1)
-        count = dmr.get_distinct_count(t1, t2)
+        count = dmr._get_distinct_count(t1, t2)
         self.assertEqual(count, 2)
         # Now we add to a newer bucket, but still within range
         self._create_test_data_points("three", b2)
-        count = dmr.get_distinct_count(t1, t2)
+        count = dmr._get_distinct_count(t1, t2)
         self.assertEqual(count, 3)
         # Now to a older bucket, out of range
         self._create_test_data_points("zero", b1 - 1)
-        count = dmr.get_distinct_count(t1, t2)
+        count = dmr._get_distinct_count(t1, t2)
         self.assertEqual(count, 3)
         # Now to a newer bucket, out of range
         self._create_test_data_points("infinity", b2 + 1)
-        count = dmr.get_distinct_count(t1, t2)
+        count = dmr._get_distinct_count(t1, t2)
         self.assertEqual(count, 3)
 
     def test_get_distinct_top_value(self):
@@ -409,23 +410,46 @@
         # Make sure we're using different buckets
         self.assertNotEqual(b1, b2)
         self._create_test_data_points("one", b1)
-        values = dmr.get_distinct_top_value(t1, t2)
+        values = dmr._get_distinct_top_value(t1, t2)
         self.assertEqual(values, [("one", 1)])
         self._create_test_data_points("two", b1)
-        values = dmr.get_distinct_top_value(t1, t2)
+        values = dmr._get_distinct_top_value(t1, t2)
         self.assertEqual(values, [("one", 1), ("two", 1)])
         self._create_test_data_points("one", b1)
-        values = dmr.get_distinct_top_value(t1, t2)
+        values = dmr._get_distinct_top_value(t1, t2)
         self.assertEqual(values, [("one", 2), ("two", 1)])
         # Create a third, and make "two" have 2 values so "three" is last
         self._create_test_data_points("two", b1)
         self._create_test_data_points("three", b1)
-        values = dmr.get_distinct_top_value(t1, t2)
+        values = dmr._get_distinct_top_value(t1, t2)
         self.assertEqual(values, [("one", 2), ("two", 2), ("three", 1)])
         # Only get the top 2
-        values = dmr.get_distinct_top_value(t1, t2, how_many=2)
+        values = dmr._get_distinct_top_value(t1, t2, how_many=2)
         self.assertEqual(values, [("one", 2), ("two", 2)])
 
+    def test_count_long_periods(self):
+        dmr = distinct.DistinctMetricReporter("test", dsn=self.dsn)
+        dmr.update("one")
+        dmr.update("one")
+        dmr.update("two")
+        dmr._save_bucket(dmr.bucket, dmr.get_bucket_no() - 2)
+        dmr = distinct.DistinctMetricReporter("test", dsn=self.dsn)
+        # this should be running in bg now too
+        # but run this way we ensure it has finished before the assert
+        dmr._update_longterm_distinct()
+        # value is loaded on startup
+        self.assertEquals(dmr.longterm_distinct_count, 2)
+        dmr.update("one")
+        dmr.update("three")
+        dmr._save_bucket(dmr.bucket, dmr.get_bucket_no() - 1)
+        # value is updated on save_bucket
+        self.assertEquals(dmr.longterm_distinct_count, 3)
+        # reset bucket
+        dmr.build_bucket()
+        now = dmr.wall_time_func()
+        result = dmr.flush(0, now)
+        self.assertTrue(("test.count_60day", 3, now) in result, result)
+
 
 class TestRedis(TestCase):
 

=== modified file 'twisted/plugins/distinctdbplugin.py'
--- twisted/plugins/distinctdbplugin.py	2011-12-15 20:19:11 +0000
+++ twisted/plugins/distinctdbplugin.py	2012-06-27 20:06:19 +0000
@@ -14,13 +14,14 @@
     bucket_size = None
     dsn = None
     metric_ids = None
+    longterm_distinct_days = None
 
     def build_metric(self, prefix, name, wall_time_func=None):
         return DistinctMetricReporter(name, prefix=prefix,
-                                      wall_time_func=wall_time_func,
-                                      bucket_size=self.bucket_size,
-                                      dsn=self.dsn, redis_host=self.redis_host,
-                                      redis_port=self.redis_port)
+            wall_time_func=wall_time_func, bucket_size=self.bucket_size,
+            dsn=self.dsn, redis_host=self.redis_host,
+            redis_port=self.redis_port,
+            longterm_distinct_days=self.longterm_distinct_days)
 
     def configure(self, options):
         self.section = dict(options.get("plugin_distinctdb", {}))
@@ -30,6 +31,8 @@
             self.bucket_size = ONE_DAY
 
         self.dsn = self.section.get("dsn", None)
+        self.longterm_distinct_days = int(
+            self.section.get("longterm_distinct_days", 60))
         self.redis_host = self.section.get("redis_host", None)
         self.redis_port = self.section.get("redis_port", None)
         if self.redis_port is not None: