From a1bed0ddc158b514257542023f022e1d8c437472 Mon Sep 17 00:00:00 2001 From: Matt Jia Date: Jun 12 2017 05:03:31 +0000 Subject: [PATCH 1/2] rename the config option ZEROMQ_PUBLISH to MESSAGE_BUS_PUBLISH We're not only using zmq, but also amqp or stomp. This option is really just for enabling or disabling message publishing. --- diff --git a/tests/test_app.py b/tests/test_app.py index aba4648..2d0437a 100644 --- a/tests/test_app.py +++ b/tests/test_app.py @@ -21,24 +21,26 @@ from waiverdb import app, config from flask_sqlalchemy import SignallingSession -class NoZmqConfig(config.Config): - ZEROMQ_PUBLISH = False +class DisabledMessagingConfig(config.Config): + MESSAGE_BUS_PUBLISH = False AUTH_METHOD = None + SQLALCHEMY_TRACK_MODIFICATIONS = True -class ZmqConfig(config.Config): - ZEROMQ_PUBLISH = True +class EnabledMessagedConfig(config.Config): + MESSAGE_BUS_PUBLISH = True AUTH_METHOD = None + SQLALCHEMY_TRACK_MODIFICATIONS = True @mock.patch('waiverdb.app.event.listen') -def test_register_events_no_zmq(mock_listen): - app.create_app(NoZmqConfig) +def test_disabled_messaging_should_not_register_events(mock_listen): + app.create_app(DisabledMessagingConfig) assert 0 == mock_listen.call_count @mock.patch('waiverdb.app.event.listen') -def test_register_events_zmq(mock_listen): - app.create_app(ZmqConfig) +def test_enabled_messaging_should_register_events(mock_listen): + app.create_app(EnabledMessagedConfig) mock_listen.assert_called_once_with( SignallingSession, 'after_commit', app.fedmsg_new_waiver) diff --git a/waiverdb/app.py b/waiverdb/app.py index 89a17a1..ce9a4af 100644 --- a/waiverdb/app.py +++ b/waiverdb/app.py @@ -109,7 +109,7 @@ def register_event_handlers(app): app (flask.Flask): The Flask object with the configured scoped session attached as the ``session`` attribute. """ - if app.config['ZEROMQ_PUBLISH']: + if app.config['MESSAGE_BUS_PUBLISH']: # A workaround for https://github.com/mitsuhiko/flask-sqlalchemy/pull/364 # can be removed after python-flask-sqlalchemy is upgraded to 2.2 from flask_sqlalchemy import SignallingSession diff --git a/waiverdb/config.py b/waiverdb/config.py index fd65797..3bd1ae5 100644 --- a/waiverdb/config.py +++ b/waiverdb/config.py @@ -34,7 +34,8 @@ class Config(object): AUTH_METHOD = 'OIDC' # Specify OIDC or Kerberos for authentication # Change it if the Kerberos service is not running on which the waiverdb is run. KERBEROS_HTTP_HOST = None - ZEROMQ_PUBLISH = True + # Set this to True or False to enable publishing to a message bus + MESSAGE_BUS_PUBLISH = True class ProductionConfig(Config): From 68ed69ad2b23dae1305ad4df68132acaffb73e12 Mon Sep 17 00:00:00 2001 From: Matt Jia Date: Jun 12 2017 05:03:31 +0000 Subject: [PATCH 2/2] support stomp messaging --- diff --git a/conf/settings.py.example b/conf/settings.py.example index 87c49a6..ec06621 100644 --- a/conf/settings.py.example +++ b/conf/settings.py.example @@ -7,3 +7,23 @@ JOURNAL_LOGGING = False #SHOW_DB_URI = False HOST= '0.0.0.0' PORT = 5004 +## Alternatively, you could use stomp to publish messages to a message bus. +#MESSAGE_PUBLISHER = 'stomp' +#STOMP_CONFIGS = { +# 'destination': 'topic://VirtualTopic.eng.waiverdb.waiver.new', +# 'connection': { +# 'host_and_ports': [ +# ('broker01', 61612), +# ('broker02', 61612), +# ], +# 'use_ssl': True, +# 'ssl_key_file': '/path/to/key/file', +# 'ssl_cert_file': '/path/to/cert/file', +# 'ssl_ca_certs': '/path/to/ca/certs', +# }, +# # Alternatively, you can connect STOMP server with username and password. +# 'credentials': { +# 'username': 'username', +# 'password': 'password', +# } +#} diff --git a/requirements.txt b/requirements.txt index 750ac6a..d7c5c31 100644 --- a/requirements.txt +++ b/requirements.txt @@ -14,6 +14,7 @@ systemd # packages for the unit tests pytest >= 2.4.2 mock +stomp.py # Documentation requirements sphinx diff --git a/tests/test_app.py b/tests/test_app.py index 2d0437a..6ecb5bf 100644 --- a/tests/test_app.py +++ b/tests/test_app.py @@ -43,4 +43,4 @@ def test_disabled_messaging_should_not_register_events(mock_listen): def test_enabled_messaging_should_register_events(mock_listen): app.create_app(EnabledMessagedConfig) mock_listen.assert_called_once_with( - SignallingSession, 'after_commit', app.fedmsg_new_waiver) + SignallingSession, 'after_commit', app.publish_new_waiver) diff --git a/tests/test_events.py b/tests/test_events.py index 89af138..97b9603 100644 --- a/tests/test_events.py +++ b/tests/test_events.py @@ -14,23 +14,12 @@ # GNU General Public License for more details. """This module contains tests for :mod:`waiverdb.events`.""" from __future__ import unicode_literals - -import pytest - import mock - -from waiverdb import events from waiverdb.models import Waiver -@mock.patch('waiverdb.events.fedmsg', None) -def test_fedmsg_new_waiver_missing_fedmsg(): - with pytest.raises(RuntimeError): - events.fedmsg_new_waiver(None) - - @mock.patch('waiverdb.events.fedmsg') -def test_fedmsg_new_waiver(mock_fedmsg, session): +def test_publish_new_waiver_with_fedmsg(mock_fedmsg, session): waiver = Waiver( result_id=1, username='jcline', diff --git a/waiverdb.spec b/waiverdb.spec index 9dd42a2..a11e259 100644 --- a/waiverdb.spec +++ b/waiverdb.spec @@ -26,6 +26,7 @@ BuildRequires: python-mock BuildRequires: pytest BuildRequires: fedmsg BuildRequires: python-flask-oidc +BuildRequires: stomppy %{?systemd_requires} BuildRequires: systemd BuildArch: noarch @@ -42,6 +43,7 @@ Requires: python2-systemd Requires: python-mock Requires: fedmsg Requires: python-flask-oidc +Requires: stomppy %description WaiverDB is a companion service to ResultsDB, for recording waivers diff --git a/waiverdb/app.py b/waiverdb/app.py index ce9a4af..130af91 100644 --- a/waiverdb/app.py +++ b/waiverdb/app.py @@ -15,7 +15,7 @@ import urlparse from flask import Flask from sqlalchemy import event -from waiverdb.events import fedmsg_new_waiver +from waiverdb.events import publish_new_waiver from waiverdb.logger import init_logging from waiverdb.api_v1 import api_v1 from waiverdb.models import db @@ -113,4 +113,4 @@ def register_event_handlers(app): # A workaround for https://github.com/mitsuhiko/flask-sqlalchemy/pull/364 # can be removed after python-flask-sqlalchemy is upgraded to 2.2 from flask_sqlalchemy import SignallingSession - event.listen(SignallingSession, 'after_commit', fedmsg_new_waiver) + event.listen(SignallingSession, 'after_commit', publish_new_waiver) diff --git a/waiverdb/config.py b/waiverdb/config.py index 3bd1ae5..8b7a07a 100644 --- a/waiverdb/config.py +++ b/waiverdb/config.py @@ -36,6 +36,8 @@ class Config(object): KERBEROS_HTTP_HOST = None # Set this to True or False to enable publishing to a message bus MESSAGE_BUS_PUBLISH = True + # Specify fedmsg or stomp for publishing messages + MESSAGE_PUBLISHER = 'fedmsg' class ProductionConfig(Config): diff --git a/waiverdb/events.py b/waiverdb/events.py index 513055c..2e0dcbe 100644 --- a/waiverdb/events.py +++ b/waiverdb/events.py @@ -23,33 +23,31 @@ using the :func:`sqlalchemy.event.listen` function. """ from __future__ import unicode_literals -from gettext import gettext as _ import logging from flask_restful import marshal -# fedmsg is an optional dependency and may not be present -try: - import fedmsg -except ImportError: - fedmsg = None - +import fedmsg +import stomp +import json from waiverdb.fields import waiver_fields from waiverdb.models import Waiver - +from waiverdb.utils import stomp_connection +from flask import current_app _log = logging.getLogger(__name__) -def fedmsg_new_waiver(session): +def publish_new_waiver(session): """ - A post-commit event hook that emits fedmsgs. + A post-commit event hook that emits messages to a message bus. The messages + can be published by either fedmsg-hub with zmq or stomp. This event is designed to be registered with a session factory:: >>> from sqlalchemy.event import listen - >>> listen(MyScopedSession, 'after_commit', fedmsg_new_waiver) + >>> listen(MyScopedSession, 'after_commit', publish_new_waiver) - The emitted fedmsg will look like:: + The emitted message will look like:: { "username": "jcline", @@ -72,17 +70,21 @@ def fedmsg_new_waiver(session): session (sqlalchemy.orm.Session): The session that was committed to the database. This session is not active and cannot emit SQL. - Raises: - RuntimeError: If fedmsg is not installed. """ - _log.debug('The fedmsg_new_waiver SQLAlchemy event has been activated.') - if fedmsg is None: - msg = _('The application has been configured to publish fedmsgs, but ' - 'fedmsg is not installed. Please install fedmsg or remove the ' - 'fedmsg SQLAlchemy event handler.') - raise RuntimeError(msg) - - for row in session.identity_map.values(): - if isinstance(row, Waiver): - _log.debug('Publishing fedmsg for %r', row) - fedmsg.publish(topic='waiver.new', msg=marshal(row, waiver_fields)) + _log.debug('The publish_new_waiver SQLAlchemy event has been activated.') + if current_app.config['MESSAGE_PUBLISHER'] == 'stomp': + with stomp_connection() as conn: + stomp_configs = current_app.config.get('STOMP_CONFIGS') + for row in session.identity_map.values(): + if isinstance(row, Waiver): + _log.debug('Publishing a message for %r', row) + msg =json.dumps(marshal(row, waiver_fields)) + kwargs = dict(body=msg, headers={}, destination=stomp_configs['destination']) + if stomp.__version__[0] < 4: + kwargs['message'] = kwargs.pop('body') # On EL7, different sig. + conn.send(**kwargs) + else: + for row in session.identity_map.values(): + if isinstance(row, Waiver): + _log.debug('Publishing a message for %r', row) + fedmsg.publish(topic='waiver.new', msg=marshal(row, waiver_fields)) diff --git a/waiverdb/utils.py b/waiverdb/utils.py index 8436483..441b2fb 100644 --- a/waiverdb/utils.py +++ b/waiverdb/utils.py @@ -11,10 +11,12 @@ import datetime import functools -from flask import request, url_for, jsonify +import stomp +from flask import request, url_for, jsonify, current_app from flask_restful import marshal from waiverdb.fields import waiver_fields from werkzeug.exceptions import NotFound +from contextlib import contextmanager def reqparse_since(since): @@ -80,3 +82,26 @@ def jsonp(func): else: return func(*args, **kwargs) return wrapped + + +@contextmanager +def stomp_connection(): + """ + Helper function for stomp connection. + """ + if current_app.config.get('STOMP_CONFIGS'): + configs = current_app.config.get('STOMP_CONFIGS') + if 'destination' not in configs or not configs['destination']: + raise RuntimeError('stomp was configured to publish messages, ' + 'but destination is not configured in STOMP_CONFIGS') + if 'connection' not in configs or not configs['connection']: + raise RuntimeError('stomp was configured to publish messages,, ' + 'but connection is not configured in STOMP_CONFIGS') + conn = stomp.Connection(**configs['connection']) + conn.start() + conn.connect(**configs.get('credentials', {})) + yield conn + conn.disconnect() + else: + raise RuntimeError('stomp was configured to publish messages, ' + 'but STOMP_CONFIGS is not configured')