fm-orchestrator/module_build_service/messaging.py

# -*- 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 <rbean@redhat.com>
#            Matt Prahl <mprahl@redhat.com>

"""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, conf


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)

    def __json__(self):
        return dict(msg_id=self.msg_id, topic=self.topic, body=self.body)

    @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<category>' + categories_re + r')(?:(?:\.)'
             r'(?P<object>build|repo|module))?(?:(?:\.)'
             r'(?P<subobject>state|build))?(?:\.)(?P<event>change|done|end|tag)$'))
        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

            # Ignore all messages from the secondary koji instances.
            if category == 'buildsys':
                instance = msg_inner_msg.get('instance', 'primary')
                if instance != 'primary':
                    log.debug("Ignoring message from %r koji hub." % instance)
                    return

            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')

                if task_id is None:
                    log.debug("Saw a koji build change with no task_id.")
                    return

                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 == 'buildsys' and event == 'tag':
                tag = msg_inner_msg.get('tag')
                artifact = msg_inner_msg.get('name')
                msg_obj = KojiTagChange(msg_id, tag, artifact)

            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'))

            elif conf.system == category == 'copr' and object == 'build':
                copr = msg_inner_msg.get('copr')
                build = msg_inner_msg.get('build')
                status = msg_inner_msg.get('status')
                pkg = msg_inner_msg.get('pkg')
                version = msg_inner_msg.get('version')
                what = msg_inner_msg.get('what')
                msg_obj = CoprBuildEnd(msg_id, build, status, copr, pkg, version, what)

            # 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):
        if task_id is None:
            raise ValueError("KojiBuildChange with a null task_id is invalid.")
        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 KojiTagChange(BaseMessage):
    """
    A class that inherits from BaseMessage to provide a message
    object for a buildsys.tag info (in fedmsg this replaces the msg dictionary)
    :param tag: the name of tag (e.g. module-123456789-build)
    :param artifact: the name of tagged artifact (e.g. module-build-macros)
    """
    def __init__(self, msg_id, tag, artifact):
        super(KojiTagChange, self).__init__(msg_id)
        self.tag = tag
        self.artifact = artifact

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 CoprBuildEnd(KojiBuildChange):
    """ A class that inherits from KojiBuildChange to provide a message
     object for a build info from Copr

     @TODO There should be a base class for CoprBuildEnd and KojiBuildChange
     and conditions in the code should check for it's descendants instead of KojiBuildChange directly.
     In such case this class would not have to inherit from koji class

    :param msg_id: the id of the msg (e.g. 2016-SomeGUID)
    :param build_id: the id of the build (e.g. 264382)
    :param status: the new build state
    (see http://copr-backend.readthedocs.io/package/constants.html#backend.constants.BuildStatus )
    :param copr: the project name
    :param pkg: the full name of what is being built
    (e.g. mutt-kz-1.5.23.1-1.20150203.git.c8504a8a.fc21)
    :param state_reason: the optional reason as to why the state changed
    """
    def __init__(self, msg_id, build_id, status, copr, pkg, version, what=None):
        ver, rel = version.split("-", 1)
        super(CoprBuildEnd, self).__init__(
            msg_id=msg_id,
            build_id=build_id,
            task_id=build_id,
            build_new_state=status,
            build_name=pkg,
            build_version=ver,
            build_release=rel,
            state_reason=what,
        )
        self.copr = copr


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', 'copr']
    },
    'amq': {
        'publish': _amq_publish,
        'services': ['koji', 'mbs']
    },
    'in_memory': {
        'publish': _in_memory_publish,
        'services': []
    }
}