Federated Message Bus¶

Write our own broker

import fedmsg

# Yield messages as they're available from a generator
for name, endpoint, topic, msg in fedmsg.tail_messages():
    print topic, msg

$ fedmsg-tail --really-pretty

import fedmsg
fedmsg.publish(topic='testing', modname='test', msg={
    'test': "Hello World",
})

import fedmsg
fedmsg.publish(name='mybus', topic='testing', modname='test', msg={
    'test': "Hello World",
})

$ echo "Hello World." | fedmsg-logger --modname=git --topic=repo.update
$ echo '{"a": 1}' | fedmsg-logger --json-input
$ fedmsg-logger --message="This is a message."
$ fedmsg-logger --message='{"a": 1}' --json-input

org.fedoraproject.ENV.CATEGORY.OBJECT[.SUBOBJECT].EVENT

$ sudo dnf install python-fedmsg-meta-fedora-infrastructure
$ fedmsg-tail --terse

$ fedmsg-tail --really-pretty

...
error: command 'swig' failed with exit status 1
...

$ sudo dnf install python-virtualenvwrapper

$ sudo dnf install python-virtualenv libffi-devel openssl-devel \
  zeromq-devel gcc

$ git clone git://github.com/fedora-infra/fedmsg.git

$ cd fedmsg
$ mkvirtualenv fedmsg
(fedmsg)$ pip install -e .[all]

(fedmsg)$ python setup.py test

$ workon fedmsg

(fedmsg)$ fedmsg-relay

(fedmsg)$ fedmsg-tail --really-pretty

(fedmsg)$ echo "Hello, world" | fedmsg-logger

$ sudo dnf install fedmsg

#"fedora-infrastructure": [
#    "tcp://hub.fedoraproject.org:9940",
#    #"tcp://stg.fedoraproject.org:9940",
#],

$ sudo dnf install fedmsg-relay
$ sudo systemctl restart fedmsg-relay
$ sudo systemctl enable fedmsg-relay

{
  "username": "root",
  "i": 1,
  "timestamp": 1393878837,
  "msg_id": "2014-f1c49f0b-5caf-49e6-b79a-cc54bcfac602",
  "topic": "org.fedoraproject.dev.logger.log",
  "msg": {
    "log": "Hello world"
  }
}

config = dict(
    endpoints={
        "relay_outbound": [
            "tcp://hostA:4001",
        ],
    },
    relay_inbound=[
        "tcp://hostA:2003",
    ],
)

$ sudo dnf install postgresql-server python-psycopg2
$ postgresql-setup initdb

host all all 127.0.0.1/32 ident sameuser
host all all ::1/128 ident sameuser

host all all 127.0.0.1/32 trust
host all all ::1/128 trust

$ systemctl start postgresql
$ systemctl enable postgresql

$ sudo -u postgres createuser -SDRPE datanommer
$ sudo -u postgres createdb -E utf8 datanommer -O datanommer

$ sudo dnf install fedmsg-hub python-datanommer-consumer datanommer-commands

config = {
    'datanommer.enabled': True,
    'datanommer.sqlalchemy.url': 'postgresql://datanommer:password@localhost/datanommer',
}

$ datanommer-create-db

$ sudo systemctl start fedmsg-hub
$ sudo systemctl enable fedmsg-hub

$ datanommer-stats
[2014-03-03 20:34:43][    fedmsg    INFO] logger has 2 entries

$ sudo dnf install datagrepper mod_wsgi

LoadModule wsgi_module modules/mod_wsgi.so

# Static resources for the datagrepper app.
Alias /datagrepper/css /usr/lib/python2.7/site-packages/datagrepper/static/css

WSGIDaemonProcess datagrepper user=fedmsg group=fedmsg maximum-requests=50000 display-name=datagrepper processes=8 threads=4 inactivity-timeout=300
WSGISocketPrefix run/wsgi
WSGIRestrictStdout Off
WSGIRestrictSignal Off
WSGIPythonOptimize 1

WSGIScriptAlias /datagrepper /usr/share/datagrepper/apache/datagrepper.wsgi

<Directory /usr/share/datagrepper/>
  WSGIProcessGroup datagrepper
  # XXX - The syntax for this is different for different versions of apache
  Require all granted
</Directory>

$ sudo systemctl restart httpd
$ sudo systemctl enable httpd

$ sudo dnf install httpie
$ http get http://localhost/datagrepper/raw/ order==desc

>>> import fedmsg
>>> fedmsg.publish(topic='testing', modname='test', msg={
...     'test': "Hello World",
... })

>>> import fedmsg
>>> fedmsg.publish(topic='tag.update', msg={
...     'user': user,
...     'tag': tag,
... })

$ echo "Hello, world." | fedmsg-logger --topic testing
$ echo '{"foo": "bar"}' | fedmsg-logger --json-input

>>> import fedmsg
>>> import fedmsg.config

>>> # Read in the config from /etc/fedmsg.d/
>>> config = fedmsg.config.load_config()

>>> for name, endpoint, topic, msg in fedmsg.tail_messages(mute=True, **config):
...     print topic, msg  # or use fedmsg.encoding.pretty_dumps(msg)

import pygments.lexers
import pygments.formatters

import fedmsg
import fedmsg.encoding
import fedmsg.consumers

class MessageColorizer(fedmsg.consumers.FedmsgConsumer):
    topic = "*"
    jsonify = False
    config_key = 'busmon.consumers.enabled'

    def consume(self, message):
        destination_topic = "colorized-messages"

        # Just so we don't create an infinite feedback loop.
        if self.destination_topic in message.topic:
            return

        # Format the incoming message
        code = pygments.highlight(
            fedmsg.encoding.pretty_dumps(fedmsg.encoding.loads(message.body)),
            pygments.lexers.JavascriptLexer(),
            pygments.formatters.HtmlFormatter(full=False)
        ).strip()

        # Ship it!
        fedmsg.publish(
            topic=self.destination_topic,
            msg=code,
        )

setup(
    ...
    entry_points={
        'moksha.consumer': (
            'colorizer = busmon.consumers:MessageColorizer',
        ),
    },
)

#!/usr/bin/env python

import json
import pprint
import zmq

def listen_and_print():
    # You can listen to stg at "tcp://stg.fedoraproject.org:9940"
    endpoint = "tcp://hub.fedoraproject.org:9940"
    # Set this to something like org.fedoraproject.prod.compose
    topic = 'org.fedoraproject.prod.'

    ctx = zmq.Context()
    s = ctx.socket(zmq.SUB)
    s.connect(endpoint)

    s.setsockopt(zmq.SUBSCRIBE, topic)

    poller = zmq.Poller()
    poller.register(s, zmq.POLLIN)

    while True:
        evts = poller.poll()  # This blocks until a message arrives
        topic, msg = s.recv_multipart()
        print topic, pprint.pformat(json.loads(msg))

if __name__ == "__main__":
    listen_and_print()

>>> JSONEncoder().encode({"foo": ["bar", "baz"]})
'{"foo": ["bar", "baz"]}'

>>> JSONEncoder().encode({"foo": ["bar", "baz"]})
'{"foo": ["bar", "baz"]}'

>>> import fedmsg.config
>>> import fedmsg.meta
>>> config = fedmsg.config.load_config([], None)
>>> fedmsg.meta.make_processors(**config)
>>> text = fedmsg.meta.msg2repr(some_message_dict, **config)

>>> list_to_series(['a', 'b', 'c', 'd'])
'a, b, and 2 others'
>>> list_to_series(['a', 'b', 'c', 'd'], N=4, oxford_comma=False)
'a, b, c and d'

{
  'subtitle': 'relrod pushed commits to ghc and 487 other packages',
  'link': None,  # This could be something.
  'icon': 'https://that-git-logo',
  'secondary_icon': 'https://that-relrod-avatar',
  'start_time': some_timestamp,
  'end_time': some_other_timestamp,
  'human_time': '5 minutes ago',
  'usernames': ['relrod'],
  'packages': ['ghc', 'nethack', ... ],
  'topics': ['org.fedoraproject.prod.git.receive'],
  'categories': ['git'],
  'msg_ids': {
      '2014-abcde': {
          'subtitle': 'relrod pushed some commits to ghc',
          'title': 'git.receive',
          'link': 'http://...',
          'icon': 'http://...',
      },
      '2014-bcdef': {
          'subtitle': 'relrod pushed some commits to nethack',
          'title': 'git.receive',
          'link': 'http://...',
          'icon': 'http://...',
      },
  },
}

# We almost always want the fedmsg-hub to be sending messages with zmq as
# opposed to amqp or stomp.  You can send with only *one* of the messaging
# backends: zeromq or amqp or stomp.  You cannot send with two or more at
# the same time.  Here, zmq is either enabled, or it is not.  If it is not,
# see the options below for how to configure stomp or amqp.
#zmq_enabled=True,

# On the other hand, if you wanted to use STOMP *instead* of zeromq, you
# could do the following...
zmq_enabled=False,
stomp_uri='broker01.example.com:61612,broker02.example.com:61612',
stomp_heartbeat=1000,
stomp_user='username',
stomp_pass='password',
stomp_ssl_crt='/path/to/ssl.crt',
stomp_ssl_key='/path/to/ssl.key',

# This is optional.
# If present, the hub will listen only to this queue for all fedmsg consumers.
# If absent, the hub will listen to all topics declared by all fedmsg consumers.
#stomp_queue='/queue/Consumer.yourqueue',

# There's usually no point in cryptographically validating messages from
# other busses.  They likely won't bear message certificates and signatures.
# Other busses usually use TLS on the connection itself for authentication
# and authorization.
validate_signatures=False,

>>> config = dict(
...     endpoints={
...         "bodhi.app01":  [
...               "tcp://app01.phx2.fedoraproject.org:3000",
...               "tcp://app01.phx2.fedoraproject.org:3001",
...               "tcp://app01.phx2.fedoraproject.org:3002",
...               "tcp://app01.phx2.fedoraproject.org:3003",
...         ],
...     },
... )

>>> config = dict(
...     srv_endpoints=[foo.example.com]
...)

certnames={
    "bodhi.app01": "bodhi-app01.stg.phx2.fedoraproject.org",
    # ... other certname mappings may follow here.
}

>>> import socket
>>> print socket.getfqdn()

routing_policy={
    "org.fedoraproject.prod.bodhi.buildroot_override.untag": [
        "bodhi-app01.phx2.fedoraproject.org",
        "bodhi-app02.phx2.fedoraproject.org",
        "bodhi-app03.phx2.fedoraproject.org",
        "bodhi-app04.phx2.fedoraproject.org",
    ],
}

>>> config = dict(
...     irc=[
...         dict(
...             network='irc.freenode.net',
...             port=6667,
...             nickname='fedmsg-dev',
...             channel='fedora-fedmsg',
...             timeout=120,
...
...             make_pretty=True,
...             make_terse=True,
...             make_short=True,
...
...             filters=dict(
...                 topic=['koji'],
...                 body=['ralph'],
...             ),
...         ),
...     ],
... )

make_short=lambda url: requests.get('http://api.bitly.com/v3/shorten?login=YOURLOGIN&apiKey=YOURAPIKEY&longUrl=%s&format=txt' % url).text.strip()

>>> irc_color_lookup = {
...     "fas": "light blue",
...     "bodhi": "green",
...     "git": "red",
...     "tagger": "brown",
...     "wiki": "purple",
...     "logger": "orange",
...     "pkgdb": "teal",
...     "buildsys": "yellow",
...     "planet": "light green",
... }

Federated Message Bus¶

Receiving Messages with Python¶

Receiving Messages from the Shell¶

Publishing Messages with Python¶

Publishing Messages from the Shell¶

Testimonials¶

Community¶

Table of Contents¶

Overview¶

Description of the problem¶

How messaging might address the problem¶

Other benefits¶

AMQP, and 0mq¶

AMQP or “Broker? Damn near killed ‘er!”¶

0mq or “Going for Broke(rless)”¶

Service discovery¶

Namespace considerations¶

The scheme¶

Bus Topology¶

Frequently Asked Questions¶

Development¶

Using a virtualenv¶

Development Dependencies¶

Cloning the Upstream Git Repo¶

Setting up your virtualenv¶

Try out the shell commands¶

Configuration¶

Deploying fedmsg for yourself¶

The basics¶

Starting fedmsg-relay¶

Test it out¶

Branching out to two machines¶

Store all messages¶

Setting up postgres¶

Setting up datanommer¶

Querying datanommer with datagrepper¶

Outro¶

Commands¶

Console Scripts¶

fedmsg-logger¶

fedmsg-tail¶

fedmsg-dg-replay¶

fedmsg-collectd¶

fedmsg-check¶

Service Daemons¶

fedmsg-hub¶

fedmsg-relay¶

fedmsg-signing-relay¶

fedmsg-irc¶

fedmsg-gateway¶

Writing your own fedmsg commands¶

Python API: Emitting Messages¶

Python API: Consuming Messages¶

“Naive” Consuming¶

“Naive” API¶

The Hub-Consumer Approach¶

A simple example¶

An Example From “busmon”¶

FedmsgConsumer API¶

DIY - Listening with Raw zeromq¶

Message Encoding – JSON¶

SQLAlchemy Utilities¶

Cryptography and Message Signing¶

Introduction¶

Certificates¶

Routing Policy¶

Configuration¶

Module Contents¶

Replay¶

Replay queries¶

“Natural Language” Representation of Messages¶

Compatibility with Other Messaging Technologies¶

Configuration¶

Configuration¶

Glossary of Configuration Values¶

List of Message Topics¶