A stomp messaging plugin.
ClosedPublic
Actions

Authored by ralph on Apr 27 2017, 7:00 PM.

Details

Reviewers

jskladan
tflink

Group Reviewers

resultsdb

Commits

rRSDB39f6bf46f441: A stomp messaging plugin.

Summary

This adds a new messaging plugin to publish over STOMP.

Test Plan

Unfortunately, you have to set up a broker to really functionally test this. :(

Diff Detail

Repository

rRSDB resultsdb

Branch

feature/stomp-publication (branched from develop)

Lint

Lint OK

Unit

Unit Tests OK

Build Status

Buildable 1132
Build 1132: arc lint + arc unit

ralph created this revision.Apr 27 2017, 7:00 PM

Harbormaster completed building B1132: Diff 3025.Apr 27 2017, 7:00 PM

I talked to @ralph on IRC and the motivation behind this is to support the upcoming "Unified Message Bus" project in Red Hat. That uses an ActiveMQ broker which speaks STOMP and they would like to have ResultsDB interface with that broker

Ping. Any review feedback on this?

We would like to enable it internally but would rather not run a patch out of band.

kparal edited reviewers, added: resultsdb; removed: kparal.May 26 2017, 10:47 AM

Who am I to say what works, and what does not :) I guess you tested it with your usecase, so feel free to merge!

This revision is now accepted and ready to land.May 26 2017, 11:09 AM

Thanks! Although, since the move to pagure.io I don't think I have rights to land changes anymore:

❯ arc land feature/stomp-publication
 TARGET  Landing onto "develop", selected by "arc.land.onto.default" configuration.
 REMOTE  Using remote "origin", the default remote under git.
 FETCH  Fetching origin/develop...
This commit will be landed:

      - 92b3c2b A stomp messaging plugin.

Landing revision 'D1191: A stomp messaging plugin.'...
 BUILDS PASSED  Harbormaster builds for the active diff completed successfully.
 PUSHING  Pushing changes to "origin/develop".
FATAL: W any taskotron/resultsdb ralph DENIED by fallthru
(or you mis-spelled the reponame)
fatal: Could not read from remote repository.

Please make sure you have the correct access rights
and the repository exists.
Usage Exception: Push failed! Fix the error and run "arc land" again.

@tflink added you into taskotron group, please try again

ralph closed this revision.May 26 2017, 2:55 PM

Revision Contents

	Path	Coverage (All)	Coverage (Touched)
M	conf/settings.py.example (18 lines)	-	-
M	resultsdb/messaging.py (59 lines)	0%	0%
M	setup.py (1 line)	0%	-

Commit	Tree	Parents	Author	Summary	Date
22fc6faba546	c84ad8816132	144249cfc37a	Ralph Bean	A stomp messaging plugin. (Show More…)	Apr 27 2017, 6:58 PM

Diff 3025

View Options

conf/settings.py.example

Show All 17 Lines
18	# default, but you could create your own.	18	# default, but you could create your own.
19	MESSAGE_BUS_PLUGIN = 'fedmsg'	19	MESSAGE_BUS_PLUGIN = 'fedmsg'
20	# You can pass extra arguments to your message bus plugin here. For instance,	20	# You can pass extra arguments to your message bus plugin here. For instance,
21	# the fedmsg plugin expects an extra `modname` argument that can be used to	21	# the fedmsg plugin expects an extra `modname` argument that can be used to
22	# configure the topic, like this:	22	# configure the topic, like this:
23	# <topic_prefix>.<environment>.<modname>.<topic>	23	# <topic_prefix>.<environment>.<modname>.<topic>
24	# e.g. org.fedoraproject.prod.taskotron.result.new	24	# e.g. org.fedoraproject.prod.taskotron.result.new
25	MESSAGE_BUS_KWARGS = {'modname': 'resultsdb'}	25	MESSAGE_BUS_KWARGS = {'modname': 'resultsdb'}
		26
		27	## Alternatively, you could use the 'stomp' messaging plugin.
		28	#MESSAGE_BUS_PLUGIN = 'stomp'
		29	#MESSAGE_BUS_KWARGS = {
		30	# 'destination': 'topic://VirtualTopic.eng.resultsdb.result.new',
		31	# 'connection': {
		32	# 'host_and_ports': [
		33	# ('broker01', '61612'),
		34	# ('broker02', '61612'),
		35	# ],
		36	# 'use_ssl': True,
		37	# 'ssl_key_file': '/path/to/key/file',
		38	# 'ssl_cert_file': '/path/to/cert/file',
		39	# 'ssl_ca_certs': '/path/to/ca/certs',
		40	# },
		41	#}
		42
		43
26	# Specify which fields are required (in addition to those already required)	44	# Specify which fields are required (in addition to those already required)
27	# when creating result/group/testcase.	45	# when creating result/group/testcase.
28	# If you want to set some result's extra-data as required, you can do so by	46	# If you want to set some result's extra-data as required, you can do so by
29	# prepending 'data.' to the name (e.g. 'data.arch').	47	# prepending 'data.' to the name (e.g. 'data.arch').
30	REQUIRED_DATA = {	48	REQUIRED_DATA = {
31	'create_result': [],	49	'create_result': [],
32	'create_group': [],	50	'create_group': [],
33	'create_testcase': [],	51	'create_testcase': [],
34	}	52	}

View Options

resultsdb/messaging.py

Show All 12 Lines
13	# You should have received a copy of the GNU General Public License along	13		# You should have received a copy of the GNU General Public License along
14	# with this program; if not, write to the Free Software Foundation, Inc.,	14		# with this program; if not, write to the Free Software Foundation, Inc.,
15	# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.	15		# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
16	#	16		#
17	# Authors:	17		# Authors:
18	# Ralph Bean <rbean@redhat.com>	18		# Ralph Bean <rbean@redhat.com>
19		19
20	import abc	20		import abc
		21		import json
21		22
22	import pkg_resources	23		import pkg_resources
23		24
24	import fedmsg	25		import fedmsg
25		26
26	import logging	27		import logging
27	log = logging.getLogger(__name__)	28		log = logging.getLogger(__name__)
28		29
▲ Show 20 Lines • Show All 69 Lines • ▼ Show 20 Line(s)		74	def create_message(self, result, prev_result):
98		99
99	# For the v2 API	100		# For the v2 API
100	if hasattr(result, 'group'):	101		if hasattr(result, 'group'):
101	msg['msg']['result']['group_url'] = result.group.ref_url	102		msg['msg']['result']['group_url'] = result.group.ref_url
102		103
103	return msg	104		return msg
104		105
105		106
		107		class StompPlugin(MessagingPlugin):
		108		def __init__(self, **kwargs):
		109		# Ensure that we can import this at startup time.
		110		import stomp
		111		self.stomp = stomp
		112
		113		super(StompPlugin, self).__init__(**kwargs)
		114
		115		# Validate that some required config is present
		116		required = ['connection', 'destination']
		117		for attr in required:
		118		if getattr(self, attr, None) is None:
		119		raise ValueError("%r required for %r." % (attr, self))
		120
		121		def publish(self, msg):
		122		msg = json.dumps(msg)
		123		kwargs = dict(body=msg, headers={}, destination=self.destination)
		124
		125		if self.stomp.__version__[0] < 4:
		126		kwargs['message'] = kwargs.pop('body') # On EL7, different sig.
		127
		128		conn = self.stomp.Connection(**self.connection)
		129		conn.start()
		130		conn.connect(**self.credentials)
		131		try:
		132		conn.send(**kwargs)
		133		finally:
		134		conn.disconnect()
		135
		136		def create_message(self, result, prev_result):
		137		task = dict(
		138		(datum.key, datum.value)
		139		for datum in result.data
		140		if datum.key in ('item', 'type',)
		141		)
		142		task['name'] = result.testcase.name
		143		msg = {
		144		'task': task,
		145		'result': {
		146		'id': result.id,
		147		'submit_time': result.submit_time.strftime("%Y-%m-%d %H:%M:%S UTC"),
		148		'prev_outcome': prev_result.outcome if prev_result else None,
		149		'outcome': result.outcome,
		150		'log_url': result.ref_url,
		151		}
		152		}
		153
		154		# For the v1 API
		155		if hasattr(result, 'job'):
		156		msg['msg']['result']['job_url'] = result.job.ref_url
		157
		158		# For the v2 API
		159		if hasattr(result, 'group'):
		160		msg['msg']['result']['group_url'] = result.group.ref_url
		161
		162		return msg
		163
		164
106	def load_messaging_plugin(name, kwargs):	165		def load_messaging_plugin(name, kwargs):
107	""" Instantiate and return the appropriate messaging plugin. """	166		""" Instantiate and return the appropriate messaging plugin. """
108	points = pkg_resources.iter_entry_points('resultsdb.messaging.plugins')	167		points = pkg_resources.iter_entry_points('resultsdb.messaging.plugins')
109	classes = {'dummy': DummyPlugin}	168		classes = {'dummy': DummyPlugin}
110	classes.update(dict([(point.name, point.load()) for point in points]))	169		classes.update(dict([(point.name, point.load()) for point in points]))
111		170
112	log.debug("Found the following installed messaging plugin %r" % classes)	171		log.debug("Found the following installed messaging plugin %r" % classes)
113	if name not in classes:	172		if name not in classes:
Show All 10 Lines

View Options

setup.py

Show All 26 Lines		21	setup(name='resultsdb',
27	packages=['resultsdb', 'resultsdb.controllers', 'resultsdb.lib', 'resultsdb.models',	27		packages=['resultsdb', 'resultsdb.controllers', 'resultsdb.lib', 'resultsdb.models',
28	'resultsdb.serializers'],	28		'resultsdb.serializers'],
29	package_dir={'resultsdb': 'resultsdb'},	29		package_dir={'resultsdb': 'resultsdb'},
30	entry_points={	30		entry_points={
31	'console_scripts': ['resultsdb=resultsdb.cli:main'],	31		'console_scripts': ['resultsdb=resultsdb.cli:main'],
32	'resultsdb.messaging.plugins': [	32		'resultsdb.messaging.plugins': [
33	'dummy=resultsdb.messaging:DummyPlugin',	33		'dummy=resultsdb.messaging:DummyPlugin',
34	'fedmsg=resultsdb.messaging:FedmsgPlugin',	34		'fedmsg=resultsdb.messaging:FedmsgPlugin',
		35		'stomp=resultsdb.messaging:StompPlugin',
35	],	36		],
36	},	37		},
37	include_package_data=True,	38		include_package_data=True,
38	)	39		)

Diff	ID	Base	Description	Created	Lint	Unit
Base			Base
Diff 1	3025	144249c		Apr 27 2017, 7:00 PM	★	★

A stomp messaging plugin.ClosedPublicActions

Details

Diff Detail

Revision Contents

Diff 3025

conf/settings.py.example

resultsdb/messaging.py

setup.py

A stomp messaging plugin.
ClosedPublic
Actions