Merge lp:~beuno/txstatsd/distinct-plugin-webstats into lp:~txstatsd-dev/txstatsd/distinct-plugin

[Martin Albisetti]

Get distinct user stats.

Get distinct user stats.

[Lucio Torre (lucio.torre)]

1- theres no need to keep all the attributes of DistinctMetricReporter on DistinctResource
20 + def __init__(self, reporter):
21 + resource.Resource.__init__(self)
22 + self.reporter = reporter
23 + self.dsn = reporter.dsn
24 + self.prefix = reporter.prefix
25 + self.name = reporter.name
26 + self.bucket_size = reporter.bucket_size

2- t1 and t2 both fall in the same bucket! (its only ten seconds difference)

3- you should also exercise the distinct part of the query, adding a repeated value in the bucket making sure it wont add to the count

[Lucio Torre (lucio.torre)]

maybe we want the get_distinct_count function to take a DateTime object instead of a timestamp? or both?

[Sidnei da Silva (sidnei)]

review: Approve

=== modified file 'distinctdb/distinctmetric.py'
--- distinctdb/distinctmetric.py	2012-03-08 16:30:49 +0000
+++ distinctdb/distinctmetric.py	2012-03-14 18:11:18 +0000
@@ -1,11 +1,14 @@
+import json
 import time
 import threading
+from functools import partial
 
 import psycopg2
 import redis
 
 from zope.interface import implements
-from twisted.internet import reactor
+from twisted.internet import reactor, threads
+from twisted.web import server, resource
 from txstatsd.itxstatsd import IMetric
 
 ONE_MINUTE = 60
@@ -13,6 +16,42 @@
 ONE_DAY = 24 * ONE_HOUR
 
 
+class JSONMethodResource(resource.Resource):
+    """Renders the result of calling C{name} on C{target} as a json result."""
+    isLeaf = True
+
+    def __init__(self, target, name):
+        self.target = target
+        self.name = name
+
+    def _render_json_error(self, request, result):
+        """Fetch the data, make the request return"""
+        request.write(json.dumps({"error": result.getErrorMessage()}))
+        request.finish()
+
+    def _render_json_result(self, request, result):
+        """Fetch the data, make the request return"""
+        request.write(json.dumps({"result": result}))
+        request.finish()
+
+    def render_GET(self, request):
+        """Return the web request asynchronously."""
+        d = threads.deferToThread(getattr(self.target, self.name), **request.args)
+        d.addCallback(partial(self._render_json_result, request))
+        d.addErrback(partial(self._render_json_error, request))
+        return server.NOT_DONE_YET
+
+
+class DistinctResource(resource.Resource):
+    """Returns statistics about unique values."""
+
+    def __init__(self, reporter):
+        resource.Resource.__init__(self)
+        self.reporter = reporter
+        self.putChild("top", JSONMethodResource(self.reporter, "get_distinct_top_value"))
+        self.putChild("count", JSONMethodResource(self.reporter, "get_distinct_count"))
+
+
 class DistinctMetricReporter(object):
     """
     Keeps an mesurement of the distinct numbers of items seen and the times
@@ -60,6 +99,10 @@
             timestamp = self.wall_time_func()
         return int(timestamp / (self.bucket_size))
 
+    def getResource(self):
+        """Return an http resource to represent this."""
+        return DistinctResource(self)
+
     def process(self, fields):
         self.update(fields[0])
 
@@ -151,3 +194,33 @@
         for item, value in items.iteritems():
             metrics.append((self.prefix + self.name + item, value, timestamp))
         return metrics
+
+    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)
+        path = self.prefix + self.name
+        c = psycopg2.connect(self.dsn)
+        cr = c.cursor()
+        cr.execute("SELECT COUNT(DISTINCT value) FROM points "
+                   "INNER JOIN paths ON (paths.id = points.path_id) "
+                   "WHERE paths.path = %s AND bucket BETWEEN %s AND %s", (
+                       path, since_bucket, until_bucket,))
+        row = cr.fetchone()
+        return row[0]
+
+    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)
+        path = self.prefix + self.name
+        c = psycopg2.connect(self.dsn)
+        cr = c.cursor()
+        cr.execute("SELECT value, COUNT(value) AS cnt FROM points "
+                   "INNER JOIN paths ON (paths.id = points.path_id) "
+                   "WHERE paths.path = %s AND bucket BETWEEN %s AND %s"
+                   "GROUP BY value ORDER BY cnt DESC LIMIT %s", (
+                       path, since_bucket, until_bucket, how_many,))
+        rows = cr.fetchall()
+        return rows
+

=== modified file 'distinctdb/tests/test_distinct.py'
--- distinctdb/tests/test_distinct.py	2012-03-08 16:30:49 +0000
+++ distinctdb/tests/test_distinct.py	2012-03-14 18:11:18 +0000
@@ -1,14 +1,17 @@
 # Copyright (C) 2011 Canonical
 # All Rights Reserved
 
+import json
 import ConfigParser
 from cStringIO import StringIO
 import os
 import time
+
 try:
     from subprocess import check_output
 except ImportError:
     import subprocess
+
     def check_output(args):
         return subprocess.Popen(args,
             stdout=subprocess.PIPE).communicate()[0]
@@ -20,12 +23,32 @@
 from twisted.internet import reactor
 from twisted.plugin import getPlugins
 from twisted.plugins import distinctdbplugin
+from twisted.web.test.test_web import DummyRequest
+
 from txstatsd.itxstatsd import IMetricFactory
 from txstatsd import service
 
 from distinctdb import distinctmetric as distinct
 
 
+class DummyReporter(object):
+
+    def __init__(self):
+        self.called = []
+
+    def get_foo(self):
+        self.called.append(("get_foo", ()))
+        return "foo"
+
+    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):
+        self.called.append(("get_distinct_top_value", (since, until, how_many)))
+        return [("one", 1), ("two", 1)]    
+
+
 class TestDistinctMetricReporter(TestCase):
 
     def test_get_bucket_no(self):
@@ -101,6 +124,53 @@
         self.assertEquals(dmr.dsn, "dbdsn")
 
 
+class TestJSONMethodResource(TestCase):
+
+    def test_render_result_as_json(self):
+        reporter = DummyReporter()
+        request = DummyRequest([])
+        resource = distinct.JSONMethodResource(reporter, "get_foo")
+        def check(result):
+            self.assertEquals({"result": "foo"},
+                              json.loads("".join(request.written)))
+        d = request.notifyFinish()
+        d.addCallback(check)
+        d.addErrback(self.fail)
+        request.render(resource)
+        return d
+
+
+class TestDistinctResource(TestCase):
+
+    def test_render_top_resource(self):
+        reporter = DummyReporter()
+        request = DummyRequest([])
+        request.args = {"since": time.time(), "until": time.time() + 1}
+        resource = distinct.DistinctResource(reporter)
+        child_resource = resource.getChildWithDefault("top", request)
+        def check(result):
+            self.assertEquals(json.dumps({"result": [("one", 1), ("two", 1)]}),
+                              "".join(request.written))
+        d = request.notifyFinish()
+        d.addCallback(check)
+        request.render(child_resource)
+        return d
+
+    def test_render_count_resource(self):
+        reporter = DummyReporter()
+        request = DummyRequest([])
+        request.args = {"since": time.time(), "until": time.time() + 1}
+        resource = distinct.DistinctResource(reporter)
+        child_resource = resource.getChildWithDefault("count", request)
+        def check(result):
+            self.assertEquals(json.dumps({"result": 42}),
+                              "".join(request.written))
+        d = request.notifyFinish()
+        d.addCallback(check)
+        request.render(child_resource)
+        return d
+
+
 class TestPlugin(TestCase):
 
     def test_factory(self):
@@ -123,6 +193,15 @@
         cr.execute("DELETE FROM points")
         cr.execute("commit")
 
+    def _create_test_data_points(self, name, bucket_no):
+        """Easily create test data."""
+        dmr = distinct.DistinctMetricReporter("test", dsn=self.dsn)
+        dmr.update(name)
+        dmr._save_bucket(dmr.bucket, bucket_no)
+
+
+class TestDatabaseMetricStorage(TestDatabase):
+
     def test_connect(self):
         cr = self.conn.cursor()
         cr.execute("SELECT 0")
@@ -176,6 +255,67 @@
         dmr2._save_bucket({}, 0)
         self.assertEquals(dmr.metric_id, dmr2.metric_id)
 
+    def test_get_distinct_count(self):
+        """Test get_distinct_count."""
+        dmr = distinct.DistinctMetricReporter("test", dsn=self.dsn)
+        t1 = time.time()
+        # add a day
+        t2 = time.time() + 60 * 60 * 24 + 1
+        b1 = dmr.get_bucket_no(t1)
+        b2 = dmr.get_bucket_no(t2)
+        # 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)
+        self.assertEqual(count, 1)
+        self._create_test_data_points("two", b1)
+        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)
+        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)
+        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)
+        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)
+        self.assertEqual(count, 3)
+
+    def test_get_distinct_top_value(self):
+        """Test get_distinct_top_value."""
+        dmr = distinct.DistinctMetricReporter("test", dsn=self.dsn)
+        t1 = time.time()
+        # add a day
+        t2 = time.time() + 60 * 60 * 24 + 1
+        b1 = dmr.get_bucket_no(t1)
+        b2 = dmr.get_bucket_no(t2)
+        # 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)
+        self.assertEqual(values, [("one", 1)])
+        self._create_test_data_points("two", b1)
+        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)
+        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)
+        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)
+        self.assertEqual(values, [("one", 2), ("two", 2)])
+
 
 class TestRedis(TestCase):
 
@@ -246,6 +386,5 @@
         dmr1 = distinct.DistinctMetricReporter("somename")
         dmr2 = distinct.DistinctMetricReporter("othename")
         dmr3 = distinct.DistinctMetricReporter("somename")
-
         self.assertEquals(dmr1.bucket_name_for(1), dmr3.bucket_name_for(1))
         self.assertNotEquals(dmr1.bucket_name_for(1), dmr2.bucket_name_for(1))