@@ -0,0 +1,1474 @@
+import time%0A%0Afrom zope.interface import implements%0A%0Afrom twisted.internet import defer%0Afrom twisted.python import log%0A%0Afrom tensor.interfaces import ITensorSource%0Afrom tensor.objects import Source%0Afrom tensor.utils import fork%0A%0Aclass Queues(Source):%0A    %22%22%22Query llen from redis-cli%0A%0A    **Configuration arguments:**%0A%0A    :param queue: Queue name (defaults to 'celery', just because)%0A    :type queue: str.%0A    :param db: DB number%0A    :type db: int.%0A    :param clipath: Path to redis-cli (default: /usr/bin/redis-cli)%0A    :type clipath: str.%0A%0A    **Metrics:**%0A%0A    :(service_name): Queue length%0A    %22%22%22%0A    implements(ITensorSource)%0A%0A    def __init__(self, *a, **kw):%0A        Source.__init__(self, *a, **kw)%0A%0A        self.queue = self.config.get('queue', 'celery')%0A        self.db = int(self.config.get('db', 0))%0A%0A        self.clipath = self.config.get('clipath', '/usr/bin/redis-cli')%0A%0A    @defer.inlineCallbacks%0A    def get(self):%0A%0A        out, err, code = yield fork(self.clipath,%0A            args=('-n', self.db, 'llen', self.queue,)%0A        )%0A%0A        events = %5B%5D%0A        if code == 0:%0A            val = int(out.strip('%5Cn').split()%5B-1%5D)%0A%0A            defer.returnValue(%0A                self.createEvent('ok', '%25s queue length' %25 self.queue, val)%0A            )%0A%0A        else:%0A            err = 'Error running %25s: %25s' %25 (self.clipath, repr(err))%0A            log.msg(err)%0A            defer.returnValue(%0A                self.createEvent('critical', err, None)%0A            )%0A

Add a redis source