2350767616

Build Type

push

github

Committed by GitHub

Commit Message

Use only viable metrics (#36)

Run Details

562 of 678 branches covered (82.89%)

Branch coverage included in aggregate %.

9 of 12 new or added lines in 5 files covered. (75.0%)

1 existing line in 1 file now uncovered.

1747 of 2012 relevant lines covered (86.83%)

0.87 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

55.97

/greenwave/consumers/consumer.py

# SPDX-License-Identifier: GPL-2.0+
import fedmsg
import logging
import requests

import fedmsg.consumers

import greenwave.app_factory
import greenwave.decision

from greenwave.monitor import (
    decision_changed_counter,
    decision_failed_counter,
    decision_unchanged_counter,
    messaging_tx_sent_ok_counter,
    messaging_tx_failed_counter,
)
from greenwave.policies import applicable_decision_context_product_version_pairs
from greenwave.utils import right_before_this_time

try:
    import fedora_messaging.api
    import fedora_messaging.exceptions
except ImportError:
    pass

log = logging.getLogger(__name__)


def _equals_except_keys(lhs, rhs, except_keys):
    keys = lhs.keys() - except_keys
    return lhs.keys() == rhs.keys() \
        and all(lhs[key] == rhs[key] for key in keys)


def _is_decision_unchanged(old_decision, decision):
    """
    Returns true only if new decision is same as old one
    (ignores result_id values).
    """
    if old_decision is None or decision is None:
        return old_decision == decision

    requirements_keys = ('satisfied_requirements', 'unsatisfied_requirements')
    if not _equals_except_keys(old_decision, decision, requirements_keys):
        return False

    ignore_keys = ('result_id',)
    for key in requirements_keys:
        old_requirements = old_decision[key]
        requirements = decision[key]
        if len(old_requirements) != len(requirements):
            return False

        for old_requirement, requirement in zip(old_requirements, requirements):
            if not _equals_except_keys(old_requirement, requirement, ignore_keys):
                return False

    return True


class Consumer(fedmsg.consumers.FedmsgConsumer):
    """
    Base class for consumers.
    """
    config_key = 'greenwave_handler'
    hub_config_prefix = 'greenwave_consumer_'
    default_topic = 'item.new'
    monitor_labels = {'handler': 'greenwave_consumer'}

    def __init__(self, hub, *args, **kwargs):
        """
        Initialize the consumer, subscribing it to the appropriate topics.

        Args:
            hub (moksha.hub.hub.CentralMokshaHub): The hub from which this handler is consuming
                messages. It is used to look up the hub config.
        """

        prefix = hub.config.get('topic_prefix')
        env = hub.config.get('environment')
        suffix = hub.config.get(f'{self.hub_config_prefix}topic_suffix', self.default_topic)
        self.topic = ['.'.join([prefix, env, suffix])]
        self.fedmsg_config = fedmsg.config.load_config()

        config = kwargs.pop('config', None)

        super().__init__(hub, *args, **kwargs)

        self.flask_app = greenwave.app_factory.create_app(config)
        self.greenwave_api_url = self.flask_app.config['GREENWAVE_API_URL']
        log.info('Greenwave handler listening on: %s', self.topic)

    def validate(self, message):
        """
        Wraps fedmsg.consumers.FedmsgConsumer.validate() to avoid propagating
        unexpected exceptions which would cause fedmsg-hub to get stuck (stops
        processing messages but doesn't quit).
        """
        try:
            return super(Consumer, self).validate(message)
        except RuntimeWarning:  # pylint: disable=try-except-raise
            raise
        except Exception:
            log.exception('Failed to validate message: %s', message)
            raise RuntimeWarning('Unexpected exception during message validation')

    def consume(self, message):
        """
        Process the given message and take action.

        Args:
            message (munch.Munch): A fedmsg about a new item.
        """
        try:
            message = message.get('body', message)
            log.debug('Processing message "%s"', message)

            with self.flask_app.app_context():
                self._consume_message(message)
        except Exception:  # pylint: disable=broad-except
            # Disallow propagating any other exception, otherwise NACK is sent
            # and the message is scheduled to be received later. But it seems
            # these messages can be only received by other consumer (or after
            # restart) otherwise the messages can block the queue completely.
            log.exception('Unexpected exception')

    def _inc(self, messaging_counter):
        """Helper method to increase monitoring counter."""
        messaging_counter.labels(**self.monitor_labels).inc()

    def _publish_decision_update_fedmsg(self, decision):
        try:
            fedmsg.publish(topic='decision.update', msg=decision)
            self._inc(messaging_tx_sent_ok_counter)
        except Exception:
            log.exception('Error sending fedmsg message')
            self._inc(messaging_tx_failed_counter)
            raise

    def _publish_decision_update_fedora_messaging(self, decision):
        try:
            msg = fedora_messaging.api.Message(
                topic='greenwave.decision.update',
                body=decision
            )
            fedora_messaging.api.publish(msg)
            self._inc(messaging_tx_sent_ok_counter)
        except fedora_messaging.exceptions.PublishReturned as e:
            log.error(
                'Fedora Messaging broker rejected message %s: %s',
                msg.id, e)
        except fedora_messaging.exceptions.ConnectionException as e:
            log.error('Error sending message %s: %s', msg.id, e)
            self._inc(messaging_tx_failed_counter)
        except Exception:  # pylint: disable=broad-except
            log.exception('Error sending fedora-messaging message')
            self._inc(messaging_tx_failed_counter)

    def _old_and_new_decisions(self, submit_time, **request_data):
        """Returns decision before and after submit time."""
        greenwave_url = self.greenwave_api_url + '/decision'
        log.debug('querying greenwave at: %s', greenwave_url)

        try:
            decision = greenwave.decision.make_decision(request_data, self.flask_app.config)

            request_data['when'] = right_before_this_time(submit_time)
            old_decision = greenwave.decision.make_decision(request_data, self.flask_app.config)
            log.debug('old decision: %s', old_decision)
        except requests.exceptions.HTTPError as e:
            log.exception('Failed to retrieve decision for data=%s, error: %s', request_data, e)
            return None, None

        return old_decision, decision

    def _publish_decision_change(
            self,
            submit_time,
            subject,
            testcase,
            product_version,
            publish_testcase):

        policy_attributes = dict(
            subject=subject,
            testcase=testcase,
        )

        if product_version:
            policy_attributes['product_version'] = product_version

        policies = self.flask_app.config['policies']
        contexts_product_versions = applicable_decision_context_product_version_pairs(
            policies, **policy_attributes)

        for decision_context, product_version in sorted(contexts_product_versions):
            old_decision, decision = self._old_and_new_decisions(
                submit_time,
                decision_context=decision_context,
                product_version=product_version,
                subject_type=subject.type,
                subject_identifier=subject.identifier,
            )
            if decision is None:
                self._inc(decision_failed_counter.labels(decision_context=decision_context))
                continue

            if _is_decision_unchanged(old_decision, decision):
                log.debug('Decision unchanged: %s', decision)
                self._inc(decision_unchanged_counter.labels(decision_context=decision_context))
                continue

            self._inc(decision_changed_counter.labels(decision_context=decision_context))

            decision.update({
                'subject_type': subject.type,
                'subject_identifier': subject.identifier,
                # subject is for backwards compatibility only:
                'subject': [subject.to_dict()],
                'decision_context': decision_context,
                'product_version': product_version,
                'previous': old_decision,
            })
            if publish_testcase:
                decision['testcase'] = testcase

            log.info('Publishing a decision update message: %r', decision)
            if self.flask_app.config['MESSAGING'] == 'fedmsg':
                self._publish_decision_update_fedmsg(decision)
            elif self.flask_app.config['MESSAGING'] == 'fedora-messaging':
                self._publish_decision_update_fedora_messaging(decision)

1	# SPDX-License-Identifier: GPL-2.0+
2	import fedmsg	1✔
3	import logging	1✔
4	import requests	1✔
5
6	import fedmsg.consumers	1✔
7
8	import greenwave.app_factory	1✔
9	import greenwave.decision	1✔
10
11	from greenwave.monitor import (	1✔
12	decision_changed_counter,
13	decision_failed_counter,
14	decision_unchanged_counter,
15	messaging_tx_sent_ok_counter,
16	messaging_tx_failed_counter,
17	)
18	from greenwave.policies import applicable_decision_context_product_version_pairs	1✔
19	from greenwave.utils import right_before_this_time	1✔
20
21	try:	1✔
22	import fedora_messaging.api	1✔
23	import fedora_messaging.exceptions	1✔
24	except ImportError:	×
25	pass	×
26
27	log = logging.getLogger(__name__)	1✔
28
29
30	def _equals_except_keys(lhs, rhs, except_keys):	1✔
31	keys = lhs.keys() - except_keys	×
32	return lhs.keys() == rhs.keys() \	×
33	and all(lhs[key] == rhs[key] for key in keys)
34
35
36	def _is_decision_unchanged(old_decision, decision):	1✔
37	"""
38	Returns true only if new decision is same as old one
39	(ignores result_id values).
40	"""
41	if old_decision is None or decision is None:	1!
42	return old_decision == decision	1✔
43
44	requirements_keys = ('satisfied_requirements', 'unsatisfied_requirements')	×
45	if not _equals_except_keys(old_decision, decision, requirements_keys):	×
46	return False	×
47
48	ignore_keys = ('result_id',)	×
49	for key in requirements_keys:	×
50	old_requirements = old_decision[key]	×
51	requirements = decision[key]	×
52	if len(old_requirements) != len(requirements):	×
53	return False	×
54
55	for old_requirement, requirement in zip(old_requirements, requirements):	×
56	if not _equals_except_keys(old_requirement, requirement, ignore_keys):	×
57	return False	×
58
59	return True	×
60
61
62	class Consumer(fedmsg.consumers.FedmsgConsumer):	1✔
63	"""
64	Base class for consumers.
65	"""
66	config_key = 'greenwave_handler'	1✔
67	hub_config_prefix = 'greenwave_consumer_'	1✔
68	default_topic = 'item.new'	1✔
69	monitor_labels = {'handler': 'greenwave_consumer'}	1✔
70
71	def __init__(self, hub, args, *kwargs):	1✔
72	"""
73	Initialize the consumer, subscribing it to the appropriate topics.
74
75	Args:
76	hub (moksha.hub.hub.CentralMokshaHub): The hub from which this handler is consuming
77	messages. It is used to look up the hub config.
78	"""
79
80	prefix = hub.config.get('topic_prefix')	1✔
81	env = hub.config.get('environment')	1✔
82	suffix = hub.config.get(f'{self.hub_config_prefix}topic_suffix', self.default_topic)	1✔
83	self.topic = ['.'.join([prefix, env, suffix])]	1✔
84	self.fedmsg_config = fedmsg.config.load_config()	1✔
85
86	config = kwargs.pop('config', None)	1✔
87
88	super().__init__(hub, args, *kwargs)	1✔
89
90	self.flask_app = greenwave.app_factory.create_app(config)	1✔
91	self.greenwave_api_url = self.flask_app.config['GREENWAVE_API_URL']	1✔
92	log.info('Greenwave handler listening on: %s', self.topic)	1✔
93
94	def validate(self, message):	1✔
95	"""
96	Wraps fedmsg.consumers.FedmsgConsumer.validate() to avoid propagating
97	unexpected exceptions which would cause fedmsg-hub to get stuck (stops
98	processing messages but doesn't quit).
99	"""
100	try:	×
101	return super(Consumer, self).validate(message)	×
102	except RuntimeWarning: # pylint: disable=try-except-raise	×
103	raise	×
104	except Exception:	×
105	log.exception('Failed to validate message: %s', message)	×
106	raise RuntimeWarning('Unexpected exception during message validation')	×
107
108	def consume(self, message):	1✔
109	"""
110	Process the given message and take action.
111
112	Args:
113	message (munch.Munch): A fedmsg about a new item.
114	"""
115	try:	1✔
116	message = message.get('body', message)	1✔
117	log.debug('Processing message "%s"', message)	1✔
118
119	with self.flask_app.app_context():	1✔
120	self._consume_message(message)	1✔
121	except Exception: # pylint: disable=broad-except	×
122	# Disallow propagating any other exception, otherwise NACK is sent
123	# and the message is scheduled to be received later. But it seems
124	# these messages can be only received by other consumer (or after
125	# restart) otherwise the messages can block the queue completely.
126	log.exception('Unexpected exception')	×
127
128	def _inc(self, messaging_counter):	1✔
129	"""Helper method to increase monitoring counter."""
130	messaging_counter.labels(**self.monitor_labels).inc()	1✔
131
132	def _publish_decision_update_fedmsg(self, decision):	1✔
133	try:	1✔
134	fedmsg.publish(topic='decision.update', msg=decision)	1✔
135	self._inc(messaging_tx_sent_ok_counter)	1✔
136	except Exception:	×
137	log.exception('Error sending fedmsg message')	×
138	self._inc(messaging_tx_failed_counter)	×
139	raise	×
140
141	def _publish_decision_update_fedora_messaging(self, decision):	1✔
142	try:	1✔
143	msg = fedora_messaging.api.Message(	1✔
144	topic='greenwave.decision.update',
145	body=decision
146	)
147	fedora_messaging.api.publish(msg)	1✔
148	self._inc(messaging_tx_sent_ok_counter)	1✔
149	except fedora_messaging.exceptions.PublishReturned as e:	×
150	log.error(	×
151	'Fedora Messaging broker rejected message %s: %s',
152	msg.id, e)
153	except fedora_messaging.exceptions.ConnectionException as e:	×
154	log.error('Error sending message %s: %s', msg.id, e)	×
155	self._inc(messaging_tx_failed_counter)	×
156	except Exception: # pylint: disable=broad-except	×
157	log.exception('Error sending fedora-messaging message')	×
158	self._inc(messaging_tx_failed_counter)	×
159
160	def _old_and_new_decisions(self, submit_time, **request_data):	1✔
161	"""Returns decision before and after submit time."""
162	greenwave_url = self.greenwave_api_url + '/decision'	1✔
163	log.debug('querying greenwave at: %s', greenwave_url)	1✔
164
165	try:	1✔
166	decision = greenwave.decision.make_decision(request_data, self.flask_app.config)	1✔
167
168	request_data['when'] = right_before_this_time(submit_time)	1✔
169	old_decision = greenwave.decision.make_decision(request_data, self.flask_app.config)	1✔
170	log.debug('old decision: %s', old_decision)	1✔
171	except requests.exceptions.HTTPError as e:	×
172	log.exception('Failed to retrieve decision for data=%s, error: %s', request_data, e)	×
173	return None, None	×
174
175	return old_decision, decision	1✔
176
177	def _publish_decision_change(	1✔
178	self,
179	submit_time,
180	subject,
181	testcase,
182	product_version,
183	publish_testcase):
184
185	policy_attributes = dict(	1✔
186	subject=subject,
187	testcase=testcase,
188	)
189
190	if product_version:	1✔
191	policy_attributes['product_version'] = product_version	1✔
192
193	policies = self.flask_app.config['policies']	1✔
194	contexts_product_versions = applicable_decision_context_product_version_pairs(	1✔
195	policies, **policy_attributes)
196
197	for decision_context, product_version in sorted(contexts_product_versions):	1✔
198	old_decision, decision = self._old_and_new_decisions(	1✔
199	submit_time,
200	decision_context=decision_context,
201	product_version=product_version,
202	subject_type=subject.type,
203	subject_identifier=subject.identifier,
204	)
205	if decision is None:	1!
NEW 206	self._inc(decision_failed_counter.labels(decision_context=decision_context))	×
UNCOV 207	continue	×
208
209	if _is_decision_unchanged(old_decision, decision):	1!
210	log.debug('Decision unchanged: %s', decision)	×
NEW 211	self._inc(decision_unchanged_counter.labels(decision_context=decision_context))	×
212	continue	×
213
214	self._inc(decision_changed_counter.labels(decision_context=decision_context))	1✔
215
216	decision.update({	1✔
217	'subject_type': subject.type,
218	'subject_identifier': subject.identifier,
219	# subject is for backwards compatibility only:
220	'subject': [subject.to_dict()],
221	'decision_context': decision_context,
222	'product_version': product_version,
223	'previous': old_decision,
224	})
225	if publish_testcase:	1!
226	decision['testcase'] = testcase	×
227
228	log.info('Publishing a decision update message: %r', decision)	1✔
229	if self.flask_app.config['MESSAGING'] == 'fedmsg':	1✔
230	self._publish_decision_update_fedmsg(decision)	1✔
231	elif self.flask_app.config['MESSAGING'] == 'fedora-messaging':	1!
232	self._publish_decision_update_fedora_messaging(decision)	1✔

release-engineering / greenwave / 2350767616

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous