From 8edde38cc445ee8388011b479e8d6c1053ec7a69 Mon Sep 17 00:00:00 2001 From: Chenxiong Qi Date: Oct 16 2017 15:31:59 +0000 Subject: [PATCH 1/2] Support to send messages to UMB Original code in messaging is refactored so that single entry method publish can be called to publish message via different backend messaging brokers according to configuration in freshmaker's config file. Signed-off-by: Chenxiong Qi --- diff --git a/conf/config.py b/conf/config.py index d17181d..31a9f44 100644 --- a/conf/config.py +++ b/conf/config.py @@ -27,7 +27,6 @@ class BaseConfiguration(object): NET_RETRY_INTERVAL = 30 SYSTEM = 'koji' - MESSAGING = 'fedmsg' # or amq PDC_URL = 'http://modularity.fedorainfracloud.org:8080/rest_api/v1' PDC_INSECURE = True PDC_DEVELOP = True @@ -193,6 +192,30 @@ class BaseConfiguration(object): 'https://id.fedoraproject.org/scope/groups', ] + # Select which messaging backend will be used, that could be fedmsg, amq, + # in_memory or rhmsg. + MESSAGING = 'fedmsg' + MESSAGING_BACKENDS = { + 'fedmsg': { + 'SERVICE': 'freshmaker', + }, + 'rhmsg': { + # Brokers to connect, e.g. + # ['amqps://host:5671', 'amqps://anotherhost:5671'] + 'BROKER_URLS': [], + # Path to certificate file used to authenticate freshmaker + 'CERT_FILE': '', + # Path to private key file used to authenticate freshmaker + 'KEY_FILE': '', + # Path to trusted CA certificate bundle. + 'CA_CERT': '', + 'TOPIC_PREFIX': 'VirtualTopic.eng.freshmaker', + }, + 'in_memory': { + 'SERVICE': 'freshmaker', + } + } + class DevConfiguration(BaseConfiguration): DEBUG = True diff --git a/freshmaker/config.py b/freshmaker/config.py index 59c949a..94b3935 100644 --- a/freshmaker/config.py +++ b/freshmaker/config.py @@ -271,6 +271,10 @@ class Config(object): 'type': str, 'default': 'dogpile.cache.memory', 'desc': 'Name of dogpile.cache backend to use.'}, + 'messaging_backends': { + 'type': dict, + 'default': {}, + 'desc': 'Configuration for each supported messaging backend.'}, } def __init__(self, conf_section_obj): @@ -354,6 +358,6 @@ class Config(object): def _setifok_messaging(self, s): s = str(s) - if s not in ("fedmsg", "amq", "in_memory"): + if s not in ("fedmsg", "amq", "in_memory", "rhmsg"): raise ValueError("Unsupported messaging system.") self._messaging = s diff --git a/freshmaker/messaging.py b/freshmaker/messaging.py index c70f07c..f8f5ab4 100644 --- a/freshmaker/messaging.py +++ b/freshmaker/messaging.py @@ -25,30 +25,58 @@ """Generic messaging functions.""" -from freshmaker import log +import json + +from freshmaker import log, conf from freshmaker.events import BaseEvent -def publish(topic, msg, conf, service): +def publish(topic, msg): """ Publish a single message to a given backend, and return - :param topic: the topic of the message (e.g. module.state.change) - :param msg: the message contents of the message (typically JSON) - :param conf: a Config object from the class in config.py - :param service: the system that is publishing the message (e.g. mbs) - :return: + + :param str topic: the topic of the message (e.g. module.state.change) + :param dict msg: the message contents of the message (typically JSON) + :return: the value returned from underlying backend "send" method. """ try: handler = _messaging_backends[conf.messaging]['publish'] except KeyError: raise KeyError("No messaging backend found for %r" % conf.messaging) - return handler(topic, msg, conf, service) + return handler(topic, msg) -def _fedmsg_publish(topic, msg, conf, service): +def _fedmsg_publish(topic, msg): # fedmsg doesn't really need access to conf, however other backends do import fedmsg - return fedmsg.publish(topic, msg=msg, modname=service) + config = conf.messaging_backends['fedmsg'] + return fedmsg.publish(topic, msg=msg, modname=config['SERVICE']) + + +def _rhmsg_publish(topic, msg): + """Send message to Unified Message Bus + + :param str topic: the topic where message will be sent to (e.g. + images.found) + :param dict msg: the message that will be sent + """ + import proton + from rhmsg.activemq.producer import AMQProducer + + config = conf.messaging_backends['rhmsg'] + producer_config = { + 'urls': config['BROKER_URLS'], + 'certificate': config['CERT_FILE'], + 'private_key': config['KEY_FILE'], + 'trusted_certificates': config['CA_CERT'], + } + with AMQProducer(**producer_config) as producer: + topic = '{0}.{1}'.format(config['TOPIC_PREFIX'], topic) + producer.through_topic(topic) + + outgoing_msg = proton.Message() + outgoing_msg.body = json.dumps(msg) + producer.send(outgoing_msg) # A counter used for in-memory messages. @@ -56,17 +84,19 @@ _in_memory_msg_id = 0 _initial_messages = [] -def _in_memory_publish(topic, msg, conf, service): +def _in_memory_publish(topic, msg): """ Puts the message into the in memory work queue. """ # Increment the message ID. global _in_memory_msg_id _in_memory_msg_id += 1 + config = conf.messaging_backends['in_memory'] + # Create fake fedmsg from the message so we can reuse # the BaseEvent.from_fedmsg code to get the particular BaseEvent # class instance. wrapped_msg = BaseEvent.from_fedmsg( - service + "." + topic, + config['SERVICE'] + "." + topic, {"msg_id": str(_in_memory_msg_id), "msg": msg}, ) @@ -90,5 +120,8 @@ _messaging_backends = { }, 'in_memory': { 'publish': _in_memory_publish + }, + 'rhmsg': { + 'publish': _rhmsg_publish } } diff --git a/tests/test_messaging.py b/tests/test_messaging.py new file mode 100644 index 0000000..a1b45ca --- /dev/null +++ b/tests/test_messaging.py @@ -0,0 +1,144 @@ +# Copyright (c) 2017 Red Hat, Inc. +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files (the "Software"), to deal +# in the Software without restriction, including without limitation the rights +# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +# copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in all +# copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +# SOFTWARE. +# +# Written by Chenxiong Qi + + +import six +import unittest + +from mock import patch + +from freshmaker import conf +from freshmaker.messaging import publish + +try: + import rhmsg +except ImportError: + rhmsg = None + + +class TestSelectMessagingBackend(unittest.TestCase): + """Test messaging backend is selected correctly in publish method""" + + @patch('freshmaker.messaging._fedmsg_publish') + @patch('freshmaker.messaging._rhmsg_publish') + @patch('freshmaker.messaging._in_memory_publish') + def test_select_backend( + self, _in_memory_publish, _rhmsg_publish, _fedmsg_publish): + fake_msg = {'build': 'n-v-r'} + + mock_messaging_backends = { + 'fedmsg': {'publish': _fedmsg_publish}, + 'rhmsg': {'publish': _rhmsg_publish}, + 'in_memory': {'publish': _in_memory_publish}, + } + with patch.dict('freshmaker.messaging._messaging_backends', + mock_messaging_backends): + with patch.object(conf, 'messaging', new='fedmsg'): + publish('images.ready', fake_msg) + _fedmsg_publish.assert_called_once_with( + 'images.ready', fake_msg) + + with patch.object(conf, 'messaging', new='rhmsg'): + publish('images.ready', fake_msg) + _rhmsg_publish.assert_called_once_with( + 'images.ready', fake_msg) + + with patch.object(conf, 'messaging', new='in_memory'): + publish('images.ready', fake_msg) + _in_memory_publish.assert_called_once_with( + 'images.ready', fake_msg) + + def test_raise_error_if_backend_not_exists(self): + messaging_patcher = patch.object(conf, 'messaging', new='XXXX') + six.assertRaisesRegex( + self, ValueError, 'Unsupported messaging system', + messaging_patcher.start) + + +class TestPublishToFedmsg(unittest.TestCase): + """Test publish message to fedmsg using _fedmsg_publish backend""" + + @patch.object(conf, 'messaging', new='fedmsg') + @patch.object(conf, 'messaging_backends', + new={'fedmsg': {'SERVICE': 'freshmaker'}}) + @patch('fedmsg.publish') + def test_publish(self, fedmsg_publish): + fake_msg = {} + publish('images.ready', fake_msg) + + fedmsg_publish.assert_called_once_with( + 'images.ready', msg=fake_msg, modname='freshmaker') + + +@unittest.skipUnless(rhmsg, 'rhmsg is not available in Fedora yet.') +@unittest.skipIf(six.PY3, 'rhmsg has no Python 3 package so far.') +class TestPublishToRhmsg(unittest.TestCase): + """Test publish message to UMB using _rhmsg_publish backend""" + + @patch.object(conf, 'messaging', new='rhmsg') + @patch('rhmsg.activemq.producer.AMQProducer') + @patch('proton.Message') + def test_publish(self, Message, AMQProducer): + fake_msg = {} + rhmsg_config = { + 'rhmsg': { + 'BROKER_URLS': ['amqps://localhost:5671'], + 'CERT_FILE': '/path/to/cert', + 'KEY_FILE': '/path/to/key', + 'CA_CERT': '/path/to/ca-cert', + 'TOPIC_PREFIX': 'VirtualTopic.eng.freshmaker', + } + } + with patch.object(conf, 'messaging_backends', new=rhmsg_config): + publish('images.ready', fake_msg) + + AMQProducer.assert_called_with(**{ + 'urls': ['amqps://localhost:5671'], + 'certificate': '/path/to/cert', + 'private_key': '/path/to/key', + 'trusted_certificates': '/path/to/ca-cert', + }) + producer = AMQProducer.return_value.__enter__.return_value + producer.through_topic.assert_called_once_with( + 'VirtualTopic.eng.freshmaker.images.ready') + producer.send.assert_called_once_with( + Message.return_value) + + +class TestInMemoryPublish(unittest.TestCase): + """Test publish message in memory using _in_memory_publish backend""" + + @patch('freshmaker.consumer.work_queue_put') + @patch('freshmaker.events.BaseEvent.from_fedmsg') + def test_publish(self, from_fedmsg, work_queue_put): + fake_msg = {} + in_memory_config = { + 'in_memory': {'SERVICE': 'freshmaker'} + } + + with patch.object(conf, 'messaging_backends', new=in_memory_config): + publish('images.ready', fake_msg) + + from_fedmsg.assert_called_once_with( + 'freshmaker.images.ready', + {'msg_id': '1', 'msg': fake_msg}) + work_queue_put.assert_called_once_with(from_fedmsg.return_value) From a6d19ecd474a4635bc18ac4e65a9fe38b9a993a9 Mon Sep 17 00:00:00 2001 From: Chenxiong Qi Date: Oct 16 2017 15:33:35 +0000 Subject: [PATCH 2/2] Send message when rebuilt images are found Message will be sent to topic images.found once all images that are needed to be rebuilt are found (before starting to rebuild). Message body contains complete event and associated builds information, each of the builds is mapped to each image to be rebuilt. json method of Event and ArtifactBuild are modified to contain event and build URL. Signed-off-by: Chenxiong Qi --- diff --git a/conf/config.py b/conf/config.py index 31a9f44..7342082 100644 --- a/conf/config.py +++ b/conf/config.py @@ -21,6 +21,8 @@ class BaseConfiguration(object): HOST = '0.0.0.0' PORT = 5001 + SERVER_NAME = 'localhost' + DEBUG = False # Global network-related values, in seconds NET_TIMEOUT = 120 diff --git a/freshmaker/handlers/errata/errata_advisory_rpms_signed.py b/freshmaker/handlers/errata/errata_advisory_rpms_signed.py index a8563d4..95b4a59 100644 --- a/freshmaker/handlers/errata/errata_advisory_rpms_signed.py +++ b/freshmaker/handlers/errata/errata_advisory_rpms_signed.py @@ -25,9 +25,8 @@ import json import koji -from freshmaker import conf -from freshmaker import log -from freshmaker import db +from freshmaker import conf, db, log +from freshmaker import messaging from freshmaker.events import ErrataAdvisoryRPMsSignedEvent from freshmaker.events import ODCSComposeStateChangeEvent from freshmaker.handlers import BaseHandler @@ -129,6 +128,8 @@ class ErrataAdvisoryRPMsSignedHandler(BaseHandler): for url in repo_urls: log.info(" - %s", url) + messaging.publish('images.found', db_event.json()) + return [] def _fake_odcs_new_compose(self, compose_source, tag, packages=None): diff --git a/freshmaker/models.py b/freshmaker/models.py index fc84249..9daf684 100644 --- a/freshmaker/models.py +++ b/freshmaker/models.py @@ -24,12 +24,14 @@ """ SQLAlchemy Database models for the Flask app """ +import flask + from datetime import datetime from sqlalchemy.orm import (validates, relationship) from flask_login import UserMixin -from freshmaker import db, log +from freshmaker import app, db, log from freshmaker.types import ArtifactType, ArtifactBuildState from freshmaker.events import ( MBSModuleStateChangeEvent, GitModuleMetadataChangeEvent, @@ -189,13 +191,17 @@ class Event(FreshmakerBase): return "" % (self.message_id, self.event_type, self.search_key) def json(self): - return { - "id": self.id, - "message_id": self.message_id, - "search_key": self.search_key, - "event_type_id": self.event_type_id, - "builds": [b.json() for b in self.builds], - } + with app.app_context(): + event_url = flask.url_for('event', id=self.id) + db.session.add(self) + return { + "id": self.id, + "message_id": self.message_id, + "search_key": self.search_key, + "event_type_id": self.event_type_id, + "url": event_url, + "builds": [b.json() for b in self.builds], + } class EventDependency(FreshmakerBase): @@ -319,22 +325,26 @@ class ArtifactBuild(FreshmakerBase): ArtifactBuildState(self.state).name, self.event.message_id) def json(self): - return { - "id": self.id, - "name": self.name, - "original_nvr": self.original_nvr, - "rebuilt_nvr": self.rebuilt_nvr, - "type": self.type, - "type_name": ArtifactType(self.type).name, - "state": self.state, - "state_name": ArtifactBuildState(self.state).name, - "state_reason": self.state_reason, - "dep_on": self.dep_on.name if self.dep_on else None, - "time_submitted": self.time_submitted, - "time_completed": self.time_completed, - "event_id": self.event_id, - "build_id": self.build_id, - } + with app.app_context(): + build_url = flask.url_for('build', id=self.id) + db.session.add(self) + return { + "id": self.id, + "name": self.name, + "original_nvr": self.original_nvr, + "rebuilt_nvr": self.rebuilt_nvr, + "type": self.type, + "type_name": ArtifactType(self.type).name, + "state": self.state, + "state_name": ArtifactBuildState(self.state).name, + "state_reason": self.state_reason, + "dep_on": self.dep_on.name if self.dep_on else None, + "time_submitted": self.time_submitted, + "time_completed": self.time_completed, + "event_id": self.event_id, + "build_id": self.build_id, + "url": build_url, + } def get_root_dep_on(self): dep_on = self.dep_on