Merge lp:~lucio.torre/txstatsd/add-redis-to-distinct-plugin into lp:~txstatsd-dev/txstatsd/distinct-plugin

[Sidnei da Silva]

add redis support for distinct plugin

add redis support for distinct plugin

[Sidnei da Silva (sidnei)]

The contained redis config does not work with lucid's redis version. Please comment out all of the vm- options + activerehashing + hash-max options, and then it should land cleanly.

review: Approve

[Ubuntu One Server Tarmac Bot (ubuntuone-server-tarmac)]

The attempt to merge lp:~lucio.torre/txstatsd/add-redis-to-distinct-plugin into lp:~txstatsd-dev/txstatsd/distinct-plugin failed. Below is the output from the failed tests.

./bin/start-database.sh
## Starting postgres in /home/tarmac/cache/txstatsd/distinct-plugin/tmp/db1 ##
The files belonging to this database system will be owned by user "tarmac".
This user must also own the server process.

The database cluster will be initialized with locale C.
The default text search configuration will be set to "english".

fixing permissions on existing directory /home/tarmac/cache/txstatsd/distinct-plugin/tmp/db1/data ... ok
creating subdirectories ... ok
selecting default max_connections ... 100
selecting default shared_buffers ... 28MB
creating configuration files ... ok
creating template1 database in /home/tarmac/cache/txstatsd/distinct-plugin/tmp/db1/data/base/1 ... ok
initializing pg_authid ... ok
initializing dependencies ... ok
creating system views ... ok
loading system objects' descriptions ... ok
creating conversions ... ok
creating dictionaries ... ok
setting privileges on built-in objects ... ok
creating information schema ... ok
vacuuming database template1 ... ok
copying template1 to template0 ... ok
copying template1 to postgres ... ok

Success. You can now start the database server using:

    /usr/lib/postgresql/8.4/bin/postgres -D /home/tarmac/cache/txstatsd/distinct-plugin/tmp/db1/data
or
    /usr/lib/postgresql/8.4/bin/pg_ctl -D /home/tarmac/cache/txstatsd/distinct-plugin/tmp/db1/data -l logfile start

waiting for server to start.... done
server started
CREATE ROLE
To set your environment so psql will connect to this DB instance type:
    export PGHOST=/home/tarmac/cache/txstatsd/distinct-plugin/tmp/db1
## Done. ##
psql -h `pwd`/tmp/db1 -d distinct < bin/schema.sql
CREATE TABLE
CREATE TABLE
CREATE INDEX
./bin/start-redis.sh
trial distinctdb/
distinctdb.tests.test_distinct
  TestDatabase
    test_connect ... [OK]
    test_create_metric_id ... [OK]
    test_find_saved_data ... [OK]
    test_load_metric_id ... [OK]
  TestDistinctMetricReporter
    test_configure ... [OK]
    test_get_bucket_no ... [OK]
    test_max ... [OK]
    test_reports ... [OK]
  TestPlugin
    test_factory ... [OK]
  TestRedis
    test_configure ... [ERROR]
    test_connect ... [ERROR]
                                                   [ERROR]
    test_load ... [ERROR]
                                                      [ERROR]
                                                      [ERROR]
                                                      [ERROR]
          ...

Unmerged revisions

=== added file 'Makefile'
--- Makefile	1970-01-01 00:00:00 +0000
+++ Makefile	2011-12-20 17:36:02 +0000
@@ -0,0 +1,26 @@
+
+trial:
+	trial distinctdb/
+
+start-database: 
+	./bin/start-database.sh
+	psql -h `pwd`/tmp/db1 -d distinct < bin/schema.sql
+
+stop-database:
+	./bin/stop-database.sh
+
+start-redis:
+	./bin/start-redis.sh
+
+stop-redis:
+	./bin/stop-redis.sh
+
+start: start-database start-redis
+
+stop: stop-redis stop-database
+
+clean:
+	rm -rf ./tmp/
+test: start trial stop clean
+
+.PHONY: trial test start-database stop-database start-redis stop-redis start stop

=== added file 'bin/redis.conf'
--- bin/redis.conf	1970-01-01 00:00:00 +0000
+++ bin/redis.conf	2011-12-20 17:36:02 +0000
@@ -0,0 +1,213 @@
+# Redis configuration file example
+
+# Note on units: when memory size is needed, it is possible to specifiy
+# it in the usual form of 1k 5GB 4M and so forth:
+#
+# 1k => 1000 bytes
+# 1kb => 1024 bytes
+# 1m => 1000000 bytes
+# 1mb => 1024*1024 bytes
+# 1g => 1000000000 bytes
+# 1gb => 1024*1024*1024 bytes
+#
+# units are case insensitive so 1GB 1Gb 1gB are all the same.
+
+# By default Redis does not run as a daemon. Use 'yes' if you need it.
+# Note that Redis will write a pid file in /var/run/redis.pid when daemonized.
+daemonize no
+
+# When running daemonized, Redis writes a pid file in /var/run/redis.pid by
+# default. You can specify a custom pid file location here.
+# pidfile ./tmp/redis.pid
+
+# Accept connections on the specified port, default is 6379
+port 16379
+
+# If you want you can bind a single interface, if the bind option is not
+# specified all the interfaces will listen for incoming connections.
+#
+bind 127.0.0.1
+
+# Close the connection after a client is idle for N seconds (0 to disable)
+timeout 300
+
+# Set server verbosity to 'debug'
+# it can be one of:
+# debug (a lot of information, useful for development/testing)
+# verbose (many rarely useful info, but not a mess like the debug level)
+# notice (moderately verbose, what you want in production probably)
+# warning (only very important / critical messages are logged)
+loglevel verbose
+
+# Specify the log file name. Also 'stdout' can be used to force
+# Redis to log on the standard output. Note that if you use standard
+# output for logging but daemonize, logs will be sent to /dev/null
+logfile tmp/redis/redis-server.log
+
+# Set the number of databases. The default database is DB 0, you can select
+# a different one on a per-connection basis using SELECT <dbid> where
+# dbid is a number between 0 and 'databases'-1
+databases 16
+
+################################ SNAPSHOTTING  #################################
+#
+# Save the DB on disk:
+#
+#   save <seconds> <changes>
+#
+#   Will save the DB if both the given number of seconds and the given
+#   number of write operations against the DB occurred.
+#
+#   In the example below the behaviour will be to save:
+#   after 900 sec (15 min) if at least 1 key changed
+#   after 300 sec (5 min) if at least 10 keys changed
+#   after 60 sec if at least 10000 keys changed
+#
+#   Note: you can disable saving at all commenting all the "save" lines.
+
+save 900 1
+save 300 10
+save 60 10000
+
+# Compress string objects using LZF when dump .rdb databases?
+# For default that's set to 'yes' as it's almost always a win.
+# If you want to save some CPU in the saving child set it to 'no' but
+# the dataset will likely be bigger if you have compressible values or keys.
+rdbcompression yes
+
+# The filename where to dump the DB
+dbfilename dump.rdb
+
+# The working directory.
+#
+# The DB will be written inside this directory, with the filename specified
+# above using the 'dbfilename' configuration directive.
+#
+# Also the Append Only File will be created inside this directory.
+#
+# Note that you must specify a directory here, not a file name.
+dir tmp/redis
+
+################################# REPLICATION #################################
+
+# Master-Slave replication. Use slaveof to make a Redis instance a copy of
+# another Redis server. Note that the configuration is local to the slave
+# so for example it is possible to configure the slave to save the DB with a
+# different interval, or to listen to another port, and so on.
+#
+# slaveof <masterip> <masterport>
+
+# If the master is password protected (using the "requirepass" configuration
+# directive below) it is possible to tell the slave to authenticate before
+# starting the replication synchronization process, otherwise the master will
+# refuse the slave request.
+#
+# masterauth <master-password>
+
+################################## SECURITY ###################################
+
+# Require clients to issue AUTH <PASSWORD> before processing any other
+# commands.  This might be useful in environments in which you do not trust
+# others with access to the host running redis-server.
+#
+# This should stay commented out for backward compatibility and because most
+# people do not need auth (e.g. they run their own servers).
+#
+# Warning: since Redis is pretty fast an outside user can try up to
+# 150k passwords per second against a good box. This means that you should
+# use a very strong password otherwise it will be very easy to break.
+#
+# requirepass foobared
+
+################################### LIMITS ####################################
+
+# Set the max number of connected clients at the same time. By default there
+# is no limit, and it's up to the number of file descriptors the Redis process
+# is able to open. The special value '0' means no limits.
+# Once the limit is reached Redis will close all the new connections sending
+# an error 'max number of clients reached'.
+#
+# maxclients 128
+
+# Don't use more memory than the specified amount of bytes.
+# When the memory limit is reached Redis will try to remove keys with an
+# EXPIRE set. It will try to start freeing keys that are going to expire
+# in little time and preserve keys with a longer time to live.
+# Redis will also try to remove objects from free lists if possible.
+#
+# If all this fails, Redis will start to reply with errors to commands
+# that will use more memory, like SET, LPUSH, and so on, and will continue
+# to reply to most read-only commands like GET.
+#
+# WARNING: maxmemory can be a good idea mainly if you want to use Redis as a
+# 'state' server or cache, not as a real DB. When Redis is used as a real
+# database the memory usage will grow over the weeks, it will be obvious if
+# it is going to use too much memory in the long run, and you'll have the time
+# to upgrade. With maxmemory after the limit is reached you'll start to get
+# errors for write operations, and this may even lead to DB inconsistency.
+#
+# maxmemory <bytes>
+
+############################## APPEND ONLY MODE ###############################
+
+# By default Redis asynchronously dumps the dataset on disk. If you can live
+# with the idea that the latest records will be lost if something like a crash
+# happens this is the preferred way to run Redis. If instead you care a lot
+# about your data and don't want to that a single record can get lost you should
+# enable the append only mode: when this mode is enabled Redis will append
+# every write operation received in the file appendonly.aof. This file will
+# be read on startup in order to rebuild the full dataset in memory.
+#
+# Note that you can have both the async dumps and the append only file if you
+# like (you have to comment the "save" statements above to disable the dumps).
+# Still if append only mode is enabled Redis will load the data from the
+# log file at startup ignoring the dump.rdb file.
+#
+# IMPORTANT: Check the BGREWRITEAOF to check how to rewrite the append
+# log file in background when it gets too big.
+
+appendonly no
+
+# The name of the append only file (default: "appendonly.aof")
+# appendfilename appendonly.aof
+
+# The fsync() call tells the Operating System to actually write data on disk
+# instead to wait for more data in the output buffer. Some OS will really flush
+# data on disk, some other OS will just try to do it ASAP.
+#
+# Redis supports three different modes:
+#
+# no: don't fsync, just let the OS flush the data when it wants. Faster.
+# always: fsync after every write to the append only log . Slow, Safest.
+# everysec: fsync only if one second passed since the last fsync. Compromise.
+#
+# The default is "everysec" that's usually the right compromise between
+# speed and data safety. It's up to you to understand if you can relax this to
+# "no" that will will let the operating system flush the output buffer when
+# it wants, for better performances (but if you can live with the idea of
+# some data loss consider the default persistence mode that's snapshotting),
+# or on the contrary, use "always" that's very slow but a bit safer than
+# everysec.
+#
+# If unsure, use "everysec".
+
+# appendfsync always
+appendfsync everysec
+# appendfsync no
+
+############################### ADVANCED CONFIG ###############################
+
+# Glue small output buffers together in order to send small replies in a
+# single TCP packet. Uses a bit more CPU but most of the times it is a win
+# in terms of number of queries per second. Use 'yes' if unsure.
+glueoutputbuf yes
+
+################################## INCLUDES ###################################
+
+# Include one or more other config files here.  This is useful if you
+# have a standard template that goes to all redis server but also need
+# to customize a few per-server settings.  Include files can include
+# other files, so use this wisely.
+#
+# include /path/to/local.conf
+# include /path/to/other.conf

=== added file 'bin/start-redis.sh'
--- bin/start-redis.sh	1970-01-01 00:00:00 +0000
+++ bin/start-redis.sh	2011-12-20 17:36:02 +0000
@@ -0,0 +1,5 @@
+#!/bin/bash
+
+mkdir -p tmp/redis
+/sbin/start-stop-daemon --start -b -m -d . -p tmp/redis.pid --exec /usr/bin/redis-server -- `pwd`/bin/redis.conf
+

=== added file 'bin/stop-redis.sh'
--- bin/stop-redis.sh	1970-01-01 00:00:00 +0000
+++ bin/stop-redis.sh	2011-12-20 17:36:02 +0000
@@ -0,0 +1,3 @@
+#!/bin/bash
+
+/sbin/start-stop-daemon --stop -p tmp/redis.pid --exec /usr/bin/redis-server -- `pwd`/bin/redis.conf

=== modified file 'distinctdb/distinctmetric.py'
--- distinctdb/distinctmetric.py	2011-12-09 14:47:23 +0000
+++ distinctdb/distinctmetric.py	2011-12-20 17:36:02 +0000
@@ -1,12 +1,16 @@
 import time
+import threading
 
 import psycopg2
+import redis
 
 from zope.interface import implements
-from twisted.internet.threads import deferToThread
+from twisted.internet import reactor
 from txstatsd.itxstatsd import IMetric
 
-ONEDAY = 60 * 60 * 24
+ONE_MINUTE = 60
+ONE_HOUR = 60 * ONE_MINUTE
+ONE_DAY = 24 * ONE_HOUR
 
 
 class DistinctMetricReporter(object):
@@ -16,8 +20,10 @@
     """
     implements(IMetric)
 
+    periods = [5 * ONE_MINUTE, ONE_HOUR, ONE_DAY]
+
     def __init__(self, name, wall_time_func=time.time, prefix="",
-            bucket_size=ONEDAY, dsn=None):
+            bucket_size=ONE_DAY, dsn=None, redis_host=None, redis_port=None):
         """Construct a metric we expect to be periodically updated.
 
         @param name: Indicates what is being instrumented.
@@ -32,8 +38,17 @@
         self.prefix = prefix
         self.bucket_size = bucket_size
         self.dsn = dsn
+        self.redis_host = redis_host
+        if redis_port is None:
+            redis.port = 6379
+        self.redis_port = redis_port
         self.metric_id = None
         self.build_bucket()
+        self.redis_flush_lock = threading.Lock()
+        self.redis_count = {}
+
+        if redis_host != None:
+            self.redis = redis.client.Redis(host=redis_host, port=redis_port)
 
     def build_bucket(self, timestamp=None):
         self.max = 0
@@ -55,6 +70,40 @@
         if value > self.max:
             self.max = value
 
+        now = self.wall_time_func()
+        if self.redis_host is not None:
+            reactor.callInThread(self._update_count, item, now)
+
+    def bucket_name_for(self, period):
+        return "bucket_" + str(period)
+
+    def _update_count(self, value, when):
+        for period in self.periods:
+            self.redis.zadd(self.bucket_name_for(period), value, when)
+
+    def _flush_redis(self, now):
+        if self.redis_flush_lock.acquire(False) is False:
+            return
+        try:
+            for period in self.periods:
+                bucket = self.bucket_name_for(period)
+                self.redis.zremrangebyscore(bucket, 0, now - period)
+                self.redis_count[bucket] = self.redis.zcard(bucket)
+        finally:
+            self.redis_flush_lock.release()
+
+    def count(self, period):
+        return self.redis_count.get(self.bucket_name_for(period), 0)
+
+    def count_5min(self):
+        return self.count(5 * ONE_MINUTE)
+
+    def count_1hour(self):
+        return self.count(ONE_HOUR)
+
+    def count_1day(self):
+        return self.count(ONE_DAY)
+
     def _save_bucket(self, bucket, bucket_no):
         path = self.prefix + self.name
         if self.metric_id is None:
@@ -80,7 +129,7 @@
 
     def save_bucket(self, bucket, bucket_no):
         if self.dsn is not None:
-            deferToThread(self._save_bucket, bucket, bucket_no)
+            reactor.callInThread(self._save_bucket, bucket, bucket_no)
 
     def flush(self, interval, timestamp):
         current_bucket = self.get_bucket_no(timestamp)
@@ -88,9 +137,16 @@
             self.save_bucket(self.bucket, self.bucket_no)
             self.build_bucket(timestamp)
 
+        if self.redis_host is not None:
+            reactor.callInThread(self._flush_redis, timestamp)
+
         metrics = []
-        items = {".count": len(self.bucket),
-                 ".max": self.max}
+        items = {".messages": len(self.bucket),
+                 ".max": self.max,
+                 ".count_5min": self.count_5min(),
+                 ".count_1hour": self.count_1hour(),
+                 ".count_1day": self.count_1day(),
+                 }
         for item, value in items.iteritems():
             metrics.append((self.prefix + self.name + item, value, timestamp))
         return metrics

=== modified file 'distinctdb/tests/test_distinct.py'
--- distinctdb/tests/test_distinct.py	2011-12-09 20:04:06 +0000
+++ distinctdb/tests/test_distinct.py	2011-12-20 17:36:02 +0000
@@ -5,11 +5,19 @@
 from cStringIO import StringIO
 import os
 import time
-import subprocess
+try:
+    from subprocess import check_output
+except ImportError:
+    import subprocess
+    def check_output(args):
+        return subprocess.Popen(args,
+            stdout=subprocess.PIPE).communicate()[0]
 
 import psycopg2
+import redis
 
 from twisted.trial.unittest import TestCase
+from twisted.internet import reactor
 from twisted.plugin import getPlugins
 from twisted.plugins import distinctdbplugin
 from txstatsd.itxstatsd import IMetricFactory
@@ -71,8 +79,10 @@
         dmr.update("three")
         self.assertEquals(result,
             {"bucket": {"one": 2, "two": 1}, "bucket_no": 0})
-        self.assertEquals(dmr.flush(1, day),
-            [("test.max", 1, day), ("test.count", 1, day)])
+        result = dmr.flush(1, day)
+
+        self.assertTrue(("test.max", 1, day) in result)
+        self.assertTrue(("test.messages", 1, day) in result)
 
     def test_configure(self):
         class TestOptions(service.OptionsGlue):
@@ -101,7 +111,7 @@
 class TestDatabase(TestCase):
 
     def setUp(self):
-        rootdir = subprocess.check_output(["bzr", "root"]).strip()
+        rootdir = check_output(["bzr", "root"]).strip()
         dsn_file = os.path.join(rootdir, "tmp", "pg.dsn")
         self.dsn = open(dsn_file).read()
         self.conn = psycopg2.connect(self.dsn)
@@ -145,3 +155,69 @@
         dmr2 = distinct.DistinctMetricReporter("test", dsn=self.dsn)
         dmr2._save_bucket({}, 0)
         self.assertEquals(dmr.metric_id, dmr2.metric_id)
+
+
+class TestRedis(TestCase):
+
+    def setUp(self):
+        reactor._initThreadPool()
+        reactor.threadpool.start()
+
+    def tearDown(self):
+        r = redis.client.Redis(host="localhost", port=16379)
+        r.flushdb()
+        reactor.threadpool.stop()
+
+    def test_connect(self):
+        r = redis.client.Redis(host="localhost", port=16379)
+        r.ping()
+
+    def test_configure(self):
+        class TestOptions(service.OptionsGlue):
+            optParameters = [["test", "t", "default", "help"]]
+            config_section = "statsd"
+
+        o = TestOptions()
+        config_file = ConfigParser.RawConfigParser()
+        config_file.readfp(StringIO("[statsd]\n\n[plugin_distinctdb]\n"
+            "redis_host = localhost\nredis_port = 16379"))
+        o.configure(config_file)
+        dmf = distinctdbplugin.DistinctMetricFactory()
+        dmf.configure(o)
+        dmr = dmf.build_metric("foo", "bar", time.time)
+        self.assertEquals(dmr.redis_host, "localhost")
+        self.assertEquals(dmr.redis_port, 16379)
+
+    def test_usage(self):
+        dmr = distinct.DistinctMetricReporter("test",
+            redis_host="localhost", redis_port=16379)
+
+        self.assertEquals(dmr.count_1hour(), 0)
+        dmr._update_count("one", 0)
+        dmr._flush_redis(1)
+        self.assertEquals(dmr.count_1hour(), 1)
+        dmr._update_count("one", 0)
+        dmr._flush_redis(1)
+        self.assertEquals(dmr.count_1hour(), 1)
+        dmr._update_count("two", 30 * distinct.ONE_MINUTE)
+        dmr._flush_redis(30 * distinct.ONE_MINUTE)
+        self.assertEquals(dmr.count_1hour(), 2)
+        dmr._flush_redis(distinct.ONE_HOUR + 10 * distinct.ONE_MINUTE)
+        self.assertEquals(dmr.count_1hour(), 1)
+
+    def test_load(self):
+        dmr = distinct.DistinctMetricReporter("test",
+            redis_host="localhost", redis_port=16379)
+        start = time.time()
+        for i in range(10000):
+            dmr.update(str(i % 1000))
+
+        while True:
+            w = len(reactor.threadpool.working)
+            if w == 0:
+                break
+            time.sleep(0.1)
+        dmr._flush_redis(time.time())
+        duration = time.time() - start
+        self.assertEquals(dmr.count_1hour(), 1000)
+        self.assertTrue(duration <  10)

=== modified file 'twisted/plugins/distinctdbplugin.py'
--- twisted/plugins/distinctdbplugin.py	2011-12-08 21:08:38 +0000
+++ twisted/plugins/distinctdbplugin.py	2011-12-20 17:36:02 +0000
@@ -2,7 +2,7 @@
 
 from twisted.plugin import IPlugin
 from txstatsd.itxstatsd import IMetricFactory
-from distinctdb.distinctmetric import DistinctMetricReporter, ONEDAY
+from distinctdb.distinctmetric import DistinctMetricReporter, ONE_DAY
 
 
 class DistinctMetricFactory(object):
@@ -19,15 +19,20 @@
         return DistinctMetricReporter(name, prefix=prefix,
                                       wall_time_func=wall_time_func,
                                       bucket_size=self.bucket_size,
-                                      dsn=self.dsn)
+                                      dsn=self.dsn, redis_host=self.redis_host,
+                                      redis_port=self.redis_port)
 
     def configure(self, options):
         self.section = dict(options.get("plugin_distinctdb", {}))
         try:
-            self.bucket_size = int(self.section.get("bucket_size", ONEDAY))
+            self.bucket_size = int(self.section.get("bucket_size", ONE_DAY))
         except ValueError:
-            self.bucket_size = ONEDAY
+            self.bucket_size = ONE_DAY
 
         self.dsn = self.section.get("dsn", None)
+        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:
+            self.redis_port = int(self.redis_port)
 
 distinct_metric_factory = DistinctMetricFactory()