We have this rad tool in Fedora Infrastructure we use for passing around server-side messages called fedmsg. It uses zeromq behind the scenes and it is built on top of a framework Luke Macken made called Moksha (which is in turn built on top of Twisted).

To cut to the chase, I have a problem where I want to be able to measure how backlogged some of our message processing consumers are. Here's a diagram of how moksha works as things stand now:

Furthermore, here's a depiction of Twisted's own event loop; all of Moksha's code that I'll be discussing lives below in the "our code" section:

Now, when a message arrives, it is picked up by one of the backends (in our case, the zeromq one) and that message is handed of to the moksha dispatcher ("our code"). The dispatcher then hands that message to any locally registered message consumers that might want it, one after another in series (a consumer is just a Python class that defines a .consume(self, message) method). Some of these message consumers are quite fast: the datanommer consumer just stuffs the message in a postgres database for later analysis. The ircbot consumer just formats the message and sends it off to freenode (although, it throttles itself so as to not get kicked for being spammy). Other consumers take a longer time to handle individual messages. The Fedora Badges message consumer has to compare the message against a couple hundred different rules and some of those rules involve making large database queries -- not quick. The Fedora Notifications consumer has to compare the message against as many different rules and then ultimately forward the message on to irc, email or google cloud messaging for android -- not quick.

At the time of this writing, we have 2,752,890 messages in our message store which has been operating since October 2012. That averages about 4 messages per minute (quite low), but we often have relatively large spikes in volume, around 120 messages per second. How much does that backlog our consumer processes? How long does it take them to catch up? We can eyeball the logs and make guesses, but I'd really like to measure and track it.

Here's an idea. We split the moksha dispatcher into a main "enqueuing" thread and a secondary "dispatching" thread.

The logic for the enqueuer is simple: "when a message arrives, put it on the work queue". The logic for the new secondary dispatcher thread is also simple: "when I find a message in the queue, hand it off to each of my registered consumers in serial". Only when the last consumer has finished a message does the dispatcher thread then return to its work queue to get the next message. The dispatcher thread works much like it did before, but we introduce a little buffer in front of it that we can measure (with collectd, in our case).

Perhaps we can take it further. Give each consumer its own thread and work queue so they can work in parallel:

Here, the enqueuer changes: "when a message arrives, put it in each consumer queue that is registered for this kind of message." Each consumer now is managed by its own thread which picks its own messages off of its own queue and handles them as they can. The advantage here is that we can measure just how backlogged each particular consumer becomes, not just the whole hub.

Things might get tricky as some consumers might have been hacked together to share state that they shouldn't be -- I know the notifications backend does some silly stuff sharing access to the irc connection between consumers. That can be dealt with, though.

So, I dunno, good idea? Bad idea? Lemme know in #fedora-apps or #moksha on freenode.

I've been waiting on the OpenShift team to crack the WebSocket nut for a while and they finally got it back in December. To try it out, I tried to set up the Moksha Demo Dashboard on two different gears.

It wasn't too tricky. I created two OpenShift "DIY"-type apps, one for the WSGI app and another for the WebSocket server (the moksha-hub). All the work in those two repos is done in the .openshift/action_hooks directories (the code is actually just installed from PyPI). Additionally, the diy/development.ini files hold all the configuration.

It's live now at http://mokshademo-threebean.rhcloud.com/ but it's our same demo as before. Other apps still in the development pipeline should be more interesting when they arrive.

Just writing to show off how easy it is to stand up the moksha demo dashboard these days (it used to be kind of tricky).

First, install some system dependencies if you don't already have them:

sudo yum install zeromq zeromq-devel python-virtualenvwrapper

Open two terminals. In the first one run:

mkvirtualenv demo-dashboard
pip install mdemos.server mdemos.menus mdemos.metrics
wget https://raw.github.com/mokshaproject/mdemos.server/master/development.ini
paster serve --reload development.ini

And in the second one run:

workon demo-dashboard
moksha-hub

"Easy." Point your browser at http://localhost:8080/ for action.

p.s. -- In other news, I got fedmsg working with zeromq-3.2 in our staging infrastructure yesterday. It required this patch to python-txzmq That one aside, python-zmq and php-zmq "just worked" in epel-test. If you're writing zeromq code, you probably want to read this porting guide.

I've been working recently on sprucing up the fedmsg docs and the underlying Moksha docs. It seemed a natural extension of that work to put together some Moksha tutorials like this one.

Here I'll be showing you how to add a websocket-powered d3 graph of user activity to a Pyramid app using, as I said, Moksha.

$ mkvirtualenv tutorial
$ pip install pyramid
$ pip install moksha.wsgi moksha.hub

$ # tw2.d3 for our frontend component.
$ pip install tw2.d3

$ # Also, install weberror for kicks.
$ pip install weberror

$ pcreate -t alchemy tutorial
$ cd tutorial/
$ rm production.ini  # moksha-hub gets confused when this is present.
$ python setup.py develop
$ initialize_tutorial_db development.ini
$ pserve --reload development.ini

[pipeline:main]
pipeline =
    egg:WebError#evalerror
    tw2
    moksha
    tutorial

[filter:tw2]
use = egg:tw2.core#middleware

[filter:moksha]
use = egg:moksha.wsgi#middleware

##moksha.domain = live.example.com
moksha.domain = localhost

moksha.notifications = True
moksha.socket.notify = True

moksha.livesocket = True
moksha.livesocket.backend = websocket
#moksha.livesocket.reconnect_interval = 5000
moksha.livesocket.websocket.port = 9998

zmq_enabled = True
#zmq_strict = True
zmq_publish_endpoints = tcp://*:3001
zmq_subscribe_endpoints = tcp://127.0.0.1:3000,tcp://127.0.0.1:3001

[app:tutorial]
##moksha.domain = live.example.com
moksha.domain = localhost

moksha.livesocket = True
moksha.livesocket.backend = websocket
moksha.livesocket.websocket.port = 9998

zmq_enabled = True
#zmq_strict = True
zmq_publish_endpoints = tcp://*:3000
zmq_subscribe_endpoints = tcp://127.0.0.1:3000,tcp://127.0.0.1:3001

from pyramid.events import NewRequest
from pyramid.events import subscriber

from moksha.hub.hub import MokshaHub

hub = None

def hub_factory(config):
    global hub
    if not hub:
        hub = MokshaHub(config)
    return hub

@subscriber(NewRequest)
def emit_message(event):
    """ For every request made of our app, emit a message to the moksha-hub.
    Given the config from the tutorial, this will go out on port 3001.
    """

    hub = hub_factory(event.request.registry.settings)
    hub.send_message(topic="tutorial.newrequest", message={})

from tw2.d3 import TimeSeriesChart
from moksha.wsgi.widgets.api.live import LiveWidget


class UsersChart(TimeSeriesChart, LiveWidget):
    id = 'users-chart'
    topic = "tutorial.newrequest"
    onmessage = """
    tw2.store['${id}'].value++;
    """

    width = 800
    height = 150

    # Keep this many data points
    n = 200
    # Initialize to n zeros
    data = [0] * n


def get_time_series_widget(config):
    return UsersChart(
        backend=config.get('moksha.livesocket.backend', 'websocket')
    )

from pyramid.events import BeforeRender
from pyramid.threadlocal import get_current_request

from moksha.wsgi.widgets.api import get_moksha_socket

from tutorial.widgets import get_time_series_widget


@subscriber(BeforeRender)
def inject_globals(event):
    """ Before templates are rendered, make moksha front-end resources
    available in the template context.
    """
    request = get_current_request()

    # Expose these as global attrs for our templates
    event['users_widget'] = get_time_series_widget(request.registry.settings)
    event['moksha_socket'] = get_moksha_socket(request.registry.settings)

<div id="bottom">
  <div class="bottom">
    <div tal:content="structure users_widget.display()"></div>
    <div tal:content="structure moksha_socket.display()"></div>
  </div>
</div>

$ moksha-hub -v hub-config.ini

$ pserve --reload development.ini

So I'm spinning up the Fedora Messaging SIG. It's been sitting idle for years; this time we're really going to do it. We're going to put 0mq messaging hooks into bodhi, koji, pkgdb, fas, you name it.

I'm building a python module called fedmsg to wrap it all and it comes with a proposal-in-development. And if you're not excited yet, we've already 0mq enabled the Moksha Hub and achieved a ~100 times speedup over AMQP/qpid.

The point of this post is to ask about times for an IRC meeting of the Messaging SIG. Right now I'm thinking Tuesdays at 16:00-17:00 UTC in #fedora-meeting. If you're interested in helping with the effort but that's a bad time for you, let me know and we'll work it out.

I'll be out at PyCon until next Wednesday; so the first meeting won't be until March 20th at the earliest.