# -*- coding: utf-8 -*- # Copyright (c) 2016 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 Ralph Bean # Matt Prahl """Generic messaging functions.""" import json import os import re try: from inspect import signature except ImportError: from funcsigs import signature from module_build_service import log class BaseMessage(object): def __init__(self, msg_id): """ A base class to abstract messages from different backends :param msg_id: the id of the msg (e.g. 2016-SomeGUID) """ self.msg_id = msg_id # Moksha calls `consumer.validate` on messages that it receives, and # even though we have validation turned off in the config there's still # a step that tries to access `msg['body']`, `msg['topic']` and # `msg.get('topic')`. # These are here just so that the `validate` method won't raise an # exception when we push our fake messages through. # Note that, our fake message pushing has worked for a while... but the # *latest* version of fedmsg has some code that exercises the bug. I # didn't hit this until I went to test in jenkins. self.body = {} self.topic = None def __repr__(self): init_sig = signature(self.__init__) args_strs = ( "{}={!r}".format(name, getattr(self, name)) if param.default != param.empty else repr(getattr(self, name)) for name, param in init_sig.parameters.items()) return "{}({})".format(type(self).__name__, ', '.join(args_strs)) def __getitem__(self, key): """ Used to trick moksha into thinking we are a dict. """ return getattr(self, key) def __setitem__(self, key, value): """ Used to trick moksha into thinking we are a dict. """ return setattr(self, key, value) def get(self, key, value=None): """ Used to trick moksha into thinking we are a dict. """ return getattr(self, key, value) @staticmethod def from_amq(topic, msg): msg_obj = None if not hasattr(msg, 'properties'): return None # Unrelated message not identifying service origin properties = json.loads(msg.properties, encoding='utf8') service = properties.get('service') if service not in _messaging_backends['amq']['services']: log.debug('Skipping msg due service=%s which is not related (msg=%r): ' % (service, msg)) return None # This probably appies only for brew # Also wouldn't be easier to use properties? if service == 'koji': content = json.loads(msg.body, encoding='utf8')['content'] log.debug("Found koji related msg: %s" % msg) method = content['info']['method'] msg_type = content['info']['type'] if method == 'newRepo': attr = content['attribute'] state = content['info']['new'] if attr == "state" and state == "CLOSED": repo_tag = content['info']['request'] assert len(repo_tag) == 1 msg_obj = KojiRepoChange(msg.id, repo_tag[0]) elif method == 'build' and msg_type == 'TaskStateChange': attr = content['attribute'] if attr == "state": build_id = content['info']['id'] # TODO: Someone with AMQ knowledge should check if # info.id is build id or task_id here. For now I presume # it is task_id. task_id = content['info']['id'] build_state = content['new'] # These are not available before build is assigned build_name = None build_version = None build_release = None nvr_req = set(['name', 'version', 'release']) if nvr_req.issubset(set(content['info'].keys())): build_name = content['info']['name'] build_version = content['info']['version'] build_release = content['info']['release'] msg_obj = KojiBuildChange( msg.id, build_id, task_id, build_state, build_name, build_version, build_release) elif service == 'mbs': log.debug("Found mbs related msg: %s" % msg) body = json.loads(msg.body, encoding='utf8') if topic == 'module.state.change': msg_obj = MBSModule( msg.id, body['id'], body['state']) if msg_obj: return msg_obj log.debug('Skipping unrecognized message: %s' % msg) return None @staticmethod def from_fedmsg(topic, msg): """ Takes a fedmsg topic and message and converts it to a message object :param topic: the topic of the fedmsg message :param msg: the message contents from the fedmsg message :return: an object of BaseMessage descent if the message is a type that the app looks for, otherwise None is returned """ topic_categories = _messaging_backends['fedmsg']['services'] categories_re = '|'.join(map(re.escape, topic_categories)) regex_pattern = re.compile( (r'(?P' + categories_re + r')(?:\.)' r'(?Pbuild|repo|module)(?:(?:\.)' r'(?Pstate))?(?:\.)(?Pchange|done)$')) regex_results = re.search(regex_pattern, topic) if regex_results: category = regex_results.group('category') object = regex_results.group('object') subobject = regex_results.group('subobject') event = regex_results.group('event') msg_id = msg.get('msg_id') msg_inner_msg = msg.get('msg') # If there isn't a msg dict in msg then this message can be skipped if not msg_inner_msg: log.debug(('Skipping message without any content with the ' 'topic "{0}"').format(topic)) return None msg_obj = None if category == 'buildsys' and object == 'build' and \ subobject == 'state' and event == 'change': build_id = msg_inner_msg.get('build_id') task_id = msg_inner_msg.get('task_id') build_new_state = msg_inner_msg.get('new') build_name = msg_inner_msg.get('name') build_version = msg_inner_msg.get('version') build_release = msg_inner_msg.get('release') msg_obj = KojiBuildChange( msg_id, build_id, task_id, build_new_state, build_name, build_version, build_release) elif category == 'buildsys' and object == 'repo' and \ subobject is None and event == 'done': repo_tag = msg_inner_msg.get('tag') msg_obj = KojiRepoChange(msg_id, repo_tag) elif category == 'mbs' and object == 'module' and \ subobject == 'state' and event == 'change': msg_obj = MBSModule( msg_id, msg_inner_msg.get('id'), msg_inner_msg.get('state')) # If the message matched the regex and is important to the app, # it will be returned if msg_obj: return msg_obj return None class KojiBuildChange(BaseMessage): """ A class that inherits from BaseMessage to provide a message object for a build's info (in fedmsg this replaces the msg dictionary) :param msg_id: the id of the msg (e.g. 2016-SomeGUID) :param build_id: the id of the build (e.g. 264382) :param build_new_state: the new build state, this is currently a Koji integer :param build_name: the name of what is being built (e.g. golang-googlecode-tools) :param build_version: the version of the build (e.g. 6.06.06) :param build_release: the release of the build (e.g. 4.fc25) :param module_build_id: the optional id of the module_build in the database :param state_reason: the optional reason as to why the state changed """ def __init__(self, msg_id, build_id, task_id, build_new_state, build_name, build_version, build_release, module_build_id=None, state_reason=None): super(KojiBuildChange, self).__init__(msg_id) self.build_id = build_id self.task_id = task_id self.build_new_state = build_new_state self.build_name = build_name self.build_version = build_version self.build_release = build_release self.module_build_id = module_build_id self.state_reason = state_reason class KojiRepoChange(BaseMessage): """ A class that inherits from BaseMessage to provide a message object for a repo's info (in fedmsg this replaces the msg dictionary) :param msg_id: the id of the msg (e.g. 2016-SomeGUID) :param repo_tag: the repo's tag (e.g. SHADOWBUILD-f25-build) """ def __init__(self, msg_id, repo_tag): super(KojiRepoChange, self).__init__(msg_id) self.repo_tag = repo_tag class MBSModule(BaseMessage): """ A class that inherits from BaseMessage to provide a message object for a module event generated by module_build_service :param msg_id: the id of the msg (e.g. 2016-SomeGUID) :param module_build_id: the id of the module build :param module_build_state: the state of the module build """ def __init__(self, msg_id, module_build_id, module_build_state): super(MBSModule, self).__init__(msg_id) self.module_build_id = module_build_id self.module_build_state = module_build_state def publish(topic, msg, conf, service): """ 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: """ 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) def _fedmsg_publish(topic, msg, conf, service): # fedmsg doesn't really need access to conf, however other backends do import fedmsg return fedmsg.publish(topic, msg=msg, modname=service) # A counter used for in-memory messages. _in_memory_msg_id = 0 _initial_messages = [] def _in_memory_publish(topic, msg, conf, service): """ Puts the message into the in memory work queue. """ # Increment the message ID. global _in_memory_msg_id _in_memory_msg_id += 1 # Create fake fedmsg from the message so we can reuse # the BaseMessage.from_fedmsg code to get the particular BaseMessage # class instance. wrapped_msg = BaseMessage.from_fedmsg( service + "." + topic, {"msg_id": str(_in_memory_msg_id), "msg": msg}, ) # Put the message to queue. from module_build_service.scheduler.consumer import work_queue_put try: work_queue_put(wrapped_msg) except ValueError as e: log.warn("No MBSConsumer found. Shutting down? %r" % e) except AttributeError as e: # In the event that `moksha.hub._hub` hasn't yet been initialized, we # need to store messages on the side until it becomes available. # As a last-ditch effort, try to hang initial messages in the config. log.warn("Hub not initialized. Queueing on the side.") _initial_messages.append(wrapped_msg) def _amq_get_messenger(conf): import proton for attr in ('amq_private_key_file', 'amq_trusted_cert_file', 'amq_cert_file'): val = getattr(conf, attr) log.debug('Checking config.%s=%s' % (attr, val)) assert os.path.exists(val), 'config.%s=%s file does not exist' % (attr, val) for attr in ('amq_recv_addresses', 'amq_dest_address'): val = getattr(conf, attr) log.debug('Checking config.%s=%s' % (attr, val)) # list values if isinstance(val, (list, tuple)): assert val, 'config.%s is not supposed to be empty' % attr # individual urls for v in val: assert v and '://' in v, 'config.%s: value "%s" does not seem like a valid url' % (attr, val) # string values else: assert val and '://' in val, 'config.%s: value "%s" does not seem like a valid url' % (attr, val) msngr = proton.Messenger() msngr.certificate = conf.amq_cert_file msngr.private_key = conf.amq_private_key_file msngr.trusted_certificates = conf.amq_trusted_cert_file msngr.start() for url in conf.amq_recv_addresses: msngr.subscribe(url) log.debug('proton.Messenger: Subscribing to address=%s' % url) return msngr def _amq_publish(topic, msg, conf, service): import proton msngr = _amq_get_messenger(conf) message = proton.Message() message.address = conf.amq_dest_address message.subject = topic message.properties['service'] = service message.content = json.dumps(msg, ensure_ascii=False).encode('utf8') msngr.put(message) msngr.send() _messaging_backends = { 'fedmsg': { 'publish': _fedmsg_publish, 'services': ['buildsys', 'mbs'] }, 'amq': { 'publish': _amq_publish, 'services': ['koji', 'mbs'] }, 'in_memory': { 'publish': _in_memory_publish, 'services': [] } }