herlo was asking in #fedora-devel about consuming fedmsg messages from FAS. He pointed out that while there is documentation about consuming messages, there's also a lot of stuff going on. Most importantly, an introduction to the workflow is missing.

So, here's a tutorial I made in response that should walk you through it. Its a little verbose and a little rough around the edges (the line-wrapping gets wonky in places) but I think it will still serve the purpose. If it raises new questions or comments, please bring them up in the comment thread here, on the messaging-sig mailing list, or in #fedora-apps on freenode.

Static files created in that tutorial can be found in and downloaded from this gist.

So, my project for Fedora Infrastructure (fedmsg) connects around with zeromq. Way back in the Spring of this year, skvidal suggested that I "fuzz" it and see what happens, "fuzz" meaning try to cram random bits down its tubes.

I have a little pair of python scripts that I carry around in my ~/scratch/ dir to debug various zeromq situations. One of the scripts is topics_pub.py; it binds to an endpoint and publishes messages on topics. The other is topics_sub.py which connects to an endpoint and prints messages it receives to stdout.

To fuzz the subscriber, I had it connect to a garbage source built with /dev/random and netcat. In one terminal, I ran:

$ cat /dev/random | nc -l 3005

and in the other, I ran:

$ python topics_sub.py "tcp://*:3005"

... and nothing happened.

To fuzz the publisher, I hooked /dev/random up to telnet:

$ python topics_pub.py "tcp://127.0.0.1:3005"
$ cat /dev/random | telnet 127.0.0.1 3005

... and it didn't fall over. Encouraging.

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

Just got out of the Fedora Infrastructure weekly meeting. Some highlights from the development side of things:

Status app -- http://status.fedoraproject.org/ is live. Keep it open in a tab in your browser; the favicon should dynamically change when sysadmin-noc declares an outage.

pkgdb migration -- There's a plan in the works to migrate some pieces of pkgdb out to other apps where those pieces make more sense. Some of those pieces (in /packages and /tagger have hit staging).

fedora-packages has a /bugs path that we intend to replace bugz.fedoraproject.org some day. How does it look? Need more work?

fedora-tagger can now export a sqlite-db for yum like pkgdb's sqlitebuildtags did. We still need to test that it works with bodhi's masher before moving forwards with it.

Raw stream -- You can now access a raw stream of zmq messages from our staging infrastructure.

I'll have a handy works-out-of-the-box fedmsg config coming to your Fedora repo soon but in the meantime, you could try connecting to it yourself by running this script like $ python topics_sub.py tcp://stg.fedoraproject.org:9940.

You won't see much; its only broadcasting the quiet solace of our staging environment. But to make it and busmon dance, you could go hit the staging wiki, login, and make an edit.

rossdylan is working on a desktop notification app that will hook into this. Awesome.

Websockets -- (moksha, ftw) This was the one of the first things I wrote when I started work on fedmsg, and it finally made it to staging.

Check it out at https://apps.stg.fedoraproject.org/busmon/. Think of it as a web based version of #fedora-fedmsg with live regex filtering. You won't see much data. It's just displaying messages from the staging bus.

New hooks -- I have an untested fedmsg koji plugin ready to try out once staging has a builder.

There's also a new supybot plugin (supybot-fedmsg) that is ready to try out, but there's no stg instance of zodbot to do this easily. I'll wait until post-freeze. I like the code a bunch.. it is kind of magical.

You can always track the status of all the hooks at the fedmsg status page.

I ran into some wild stuff the other day while working on fedmsg for Fedora Infrastructure (FI).

tl;dr -> atexit won't fire for threads. Objects stored in an instance of threading.local do get reliably cleaned up, though. Implementing __del__ on an object there seems to be a sufficient replacement for atexit in a multi-threaded environment.

Some Context -- The goal of fedmsg is to link all the services in FI together with a message bus. We're using python (naturally) and zeromq. lmacken's Moksha is in the mix which lets us swap out zeromq for AMQP or STOMP if we like.. (we need AMQP in order to get messages from bugzilla, for instance).

We actually have a first iteration running in production now. You can see reprs of the live messages if you hop into #fedora-fedmsg on freenode.

One of the secondary goals of fedmsg (and the decisive one for this blog post) is to be as simple as possible to invoke. If we make sure that our data models have __json__ methods, then invocation can look like:

import fedmsg
fedmsg.publish(topic='event', msg=my_obj)

fedmsg does lots of "neat" stuff here. It:

• Creates a zeromq context and socket as a singleton to keep around.
• Changes the topic to org.fedoraproject.blog.event.
• Converts my_obj to a dict by calling __json__ if necessary.
• Cryptographically signs it if the config file says to do so.
• Sends it out.

The flipside of simplying the API is that we offloaded a lot of complexity to the configuration in /etc/fedmsg.d/. i.e., what port? where are the certificates? Nonetheless, I'm pretty happy with it.

The Snag -- fedmsg played nicely in multi-process environments (say, mod_wsgi). Each process would pick a different port and nobody stomped on each others resources. It wasn't until last week that we stumbled into a use case that required a multi-threaded environment. Threads fought over the zeromq socket for their process: sadness.

Solution: Hang my singleton wrapper on threading.local! I wrote some "integration-y" tests and it worked like a charm.

The last problem was cleanup. I had hitherto been using atexit.register to cleanup the zeromq context and sockets at the end of a process, but it wasn't been called when threads finished up. What to do?

First level of evil: I dove into the source of CPython (for the first time) looking for thread exit hooks. No luck.

Second level of evil: I dove into inspect. Could I rewrite the call stack and inject fedmsg cleanup code at the end of my topmost parent's frame? No luck (CPython told me "read-only").

The Solution -- Something, somewhere is either responsibly cleaning up or calling del on threading.local's members.

I had simply to add a __del__ method to my main FedmsgContext class and everything fell into place at threads' end.