From 5eefffa76d067afd71b2210810adcbe4aa7895e4 Mon Sep 17 00:00:00 2001 From: Filip Valder Date: Apr 11 2019 11:52:28 +0000 Subject: Re-work monitor module incl. messaging counters --- diff --git a/functional-tests/test_monitor.py b/functional-tests/test_monitor.py new file mode 100644 index 0000000..9562408 --- /dev/null +++ b/functional-tests/test_monitor.py @@ -0,0 +1,33 @@ +# SPDX-License-Identifier: GPL-2.0+ + +import importlib +import os +import pytest +import requests +import greenwave.monitor + + +min_num_of_metrics = 23 + + +def test_metrics(requests_session, greenwave_server): + r = requests_session.get(greenwave_server + 'api/v1.0/metrics') + + assert r.status_code == 200 + assert len([l for l in r.text.splitlines() + if l.startswith('# TYPE')]) >= min_num_of_metrics + + +def test_standalone_metrics_server_disabled_by_default(requests_session): + with pytest.raises(requests.exceptions.ConnectionError): + requests_session.get('http://127.0.0.1:10040/metrics') + + +def test_standalone_metrics_server(requests_session): + os.environ['MONITOR_STANDALONE_METRICS_SERVER_ENABLE'] = 'true' + importlib.reload(greenwave.monitor) + + r = requests_session.get('http://127.0.0.1:10040/metrics') + + assert len([l for l in r.text.splitlines() + if l.startswith('# TYPE')]) >= min_num_of_metrics diff --git a/greenwave/api_v1.py b/greenwave/api_v1.py index aac01f4..fdf9602 100644 --- a/greenwave/api_v1.py +++ b/greenwave/api_v1.py @@ -11,7 +11,7 @@ from greenwave.policies import (summarize_answers, from greenwave.resources import ResultsRetriever, retrieve_waivers from greenwave.safe_yaml import SafeYAMLError from greenwave.utils import insert_headers, jsonp -from greenwave.monitoring import ( +from greenwave.monitor import ( registry, decision_exception_counter, decision_request_duration_seconds, diff --git a/greenwave/app_factory.py b/greenwave/app_factory.py index 276fe6f..18db278 100644 --- a/greenwave/app_factory.py +++ b/greenwave/app_factory.py @@ -5,6 +5,7 @@ import shutil from flask import Flask from greenwave.api_v1 import api +from greenwave.monitor import monitor_api from greenwave.utils import json_error, load_config, sha1_mangle_key from greenwave.policies import load_policies, RemoteRule @@ -63,6 +64,7 @@ def create_app(config_obj=None): # register blueprints app.register_blueprint(api, url_prefix="/api/v1.0") + app.register_blueprint(monitor_api, url_prefix="/api/v1.0") app.add_url_rule('/healthcheck', view_func=healthcheck) # Initialize the cache. diff --git a/greenwave/consumers/fedora_messaging_consumer.py b/greenwave/consumers/fedora_messaging_consumer.py index a4d994f..72dbadf 100644 --- a/greenwave/consumers/fedora_messaging_consumer.py +++ b/greenwave/consumers/fedora_messaging_consumer.py @@ -13,6 +13,9 @@ import logging from greenwave.consumers.resultsdb import ResultsDBHandler from greenwave.consumers.waiverdb import WaiverDBHandler +from greenwave.monitor import ( + messaging_rx_counter, messaging_rx_ignored_counter, + messaging_rx_processed_ok_counter, messaging_rx_failed_counter) from fedora_messaging.config import conf @@ -41,9 +44,11 @@ def fedora_messaging_callback(message): """ log.info( 'Received message from fedora-messaging with topic: %s', message.topic) + messaging_rx_counter.inc() consumer_config = conf["consumer_config"] if message.topic.endswith("taskotron.result.new"): # New resultsdb results + messaging_rx_counter.labels(handler="resultsdb").inc() config = { "topic_prefix": consumer_config["topic_prefix"], "environment": consumer_config["environment"], @@ -53,10 +58,16 @@ def fedora_messaging_callback(message): handler = ResultsDBHandler(hub) msg = {"body": {'msg': message.body}} log.info('Sending message received to: ResultsDBHandler') - handler.consume(msg) + try: + handler.consume(msg) + messaging_rx_processed_ok_counter.labels(handler="resultsdb").inc() + except Exception: + messaging_rx_failed_counter.labels(handler="resultsdb").inc() + raise elif message.topic.endswith('waiver.new'): # New waiver submitted + messaging_rx_counter.labels(handler="waiverdb").inc() config = { "topic_prefix": consumer_config["topic_prefix"], "environment": consumer_config["environment"], @@ -66,4 +77,11 @@ def fedora_messaging_callback(message): handler = WaiverDBHandler(hub) msg = {"body": {'msg': message.body}} log.info('Sending message received to: WaiverDBHandler') - handler.consume(msg) + try: + handler.consume(msg) + messaging_rx_processed_ok_counter.labels(handler="waiverdb").inc() + except Exception: + messaging_rx_failed_counter.labels(handler="waiverdb").inc() + raise + + messaging_rx_ignored_counter.inc() diff --git a/greenwave/consumers/resultsdb.py b/greenwave/consumers/resultsdb.py index 5503d87..1194cf5 100644 --- a/greenwave/consumers/resultsdb.py +++ b/greenwave/consumers/resultsdb.py @@ -18,7 +18,10 @@ import requests import greenwave.app_factory import greenwave.resources from greenwave.api_v1 import subject_type_identifier_to_list -from greenwave.monitoring import publish_decision_exceptions_result_counter +from greenwave.monitor import ( + publish_decision_exceptions_result_counter, + messaging_tx_to_send_counter, messaging_tx_stopped_counter, + messaging_tx_sent_ok_counter, messaging_tx_failed_counter) from greenwave.policies import applicable_decision_context_product_version_pairs import xmlrpc.client @@ -276,6 +279,7 @@ class ResultsDBHandler(fedmsg.consumers.FedmsgConsumer): log.info('Getting greenwave info') for decision_context, product_version in sorted(contexts_product_versions): + messaging_tx_to_send_counter.labels(handler='resultsdb').inc() greenwave_url = self.greenwave_api_url + '/decision' data = { @@ -297,40 +301,54 @@ class ResultsDBHandler(fedmsg.consumers.FedmsgConsumer): log.debug('old decision: %s', old_decision) except requests.exceptions.HTTPError as e: log.exception('Failed to retrieve decision for data=%s, error: %s', data, e) + messaging_tx_stopped_counter.labels(handler='resultsdb').inc() continue if _is_decision_unchanged(old_decision, decision): log.debug('Skipped emitting fedmsg, decision did not change: %s', decision) - else: - decision.update({ - 'subject_type': subject_type, - 'subject_identifier': subject_identifier, - # subject is for backwards compatibility only: - 'subject': subject_type_identifier_to_list(subject_type, - subject_identifier), - 'decision_context': decision_context, - 'product_version': product_version, - 'previous': old_decision, - }) - log.info( - 'Emitted a message on the bus, %r, with the topic ' - '"greenwave.decision.update"', decision) - if self.flask_app.config['MESSAGING'] == 'fedmsg': - log.debug(' - to fedmsg') + messaging_tx_stopped_counter.labels(handler='resultsdb').inc() + continue + + decision.update({ + 'subject_type': subject_type, + 'subject_identifier': subject_identifier, + # subject is for backwards compatibility only: + 'subject': subject_type_identifier_to_list(subject_type, + subject_identifier), + 'decision_context': decision_context, + 'product_version': product_version, + 'previous': old_decision, + }) + log.info( + 'Emitting a message on the bus, %r, with the topic ' + '"greenwave.decision.update"', decision) + if self.flask_app.config['MESSAGING'] == 'fedmsg': + log.debug(' - to fedmsg') + try: fedmsg.publish(topic='decision.update', msg=decision) - elif self.flask_app.config['MESSAGING'] == 'fedora-messaging': - log.debug(' - to fedora-messaging') - try: - msg = fedora_messaging.api.Message( - topic='greenwave.decision.update', - body=decision - ) - fedora_messaging.api.publish(msg) - except fedora_messaging.exceptions.PublishReturned as e: - log.warning( - 'Fedora Messaging broker rejected message %s: %s', - msg.id, e) - except fedora_messaging.exceptions.ConnectionException as e: - log.warning('Error sending message %s: %s', msg.id, e) - except Exception: # pylint: disable=broad-except - log.exception('Error sending fedora-messaging message') + messaging_tx_sent_ok_counter.labels(handler='resultsdb').inc() + except Exception: + messaging_tx_failed_counter.labels(handler='resultsdb').inc() + raise + elif self.flask_app.config['MESSAGING'] == 'fedora-messaging': + log.debug(' - to fedora-messaging') + try: + msg = fedora_messaging.api.Message( + topic='greenwave.decision.update', + body=decision + ) + fedora_messaging.api.publish(msg) + messaging_tx_sent_ok_counter.labels(handler='resultsdb').inc() + except fedora_messaging.exceptions.PublishReturned as e: + log.warning( + 'Fedora Messaging broker rejected message %s: %s', + msg.id, e) + messaging_tx_stopped_counter.labels(handler='resultsdb').inc() + except fedora_messaging.exceptions.ConnectionException as e: + log.warning('Error sending message %s: %s', msg.id, e) + messaging_tx_failed_counter.labels(handler='resultsdb').inc() + except Exception: # pylint: disable=broad-except + log.exception('Error sending fedora-messaging message') + messaging_tx_failed_counter.labels(handler='resultsdb').inc() + + messaging_tx_stopped_counter.labels(handler='resultsdb').inc() diff --git a/greenwave/consumers/waiverdb.py b/greenwave/consumers/waiverdb.py index 282e5a2..e8b5452 100644 --- a/greenwave/consumers/waiverdb.py +++ b/greenwave/consumers/waiverdb.py @@ -17,7 +17,10 @@ import requests import greenwave.app_factory from greenwave.api_v1 import subject_type_identifier_to_list -from greenwave.monitoring import publish_decision_exceptions_waiver_counter +from greenwave.monitor import ( + publish_decision_exceptions_waiver_counter, + messaging_tx_to_send_counter, messaging_tx_stopped_counter, + messaging_tx_sent_ok_counter, messaging_tx_failed_counter) from greenwave.policies import applicable_decision_context_product_version_pairs try: @@ -98,6 +101,7 @@ class WaiverDBHandler(fedmsg.consumers.FedmsgConsumer): product_version=product_version) for decision_context, product_version in sorted(contexts_product_versions): + messaging_tx_to_send_counter.labels(handler='waiverdb').inc() data = { 'decision_context': decision_context, 'product_version': product_version, @@ -111,6 +115,7 @@ class WaiverDBHandler(fedmsg.consumers.FedmsgConsumer): if not response.ok: log.error(response.text) + messaging_tx_stopped_counter.labels(handler='waiverdb').inc() continue decision = response.json() @@ -126,44 +131,58 @@ class WaiverDBHandler(fedmsg.consumers.FedmsgConsumer): if not response.ok: log.error(response.text) + messaging_tx_stopped_counter.labels(handler='waiverdb').inc() continue old_decision = response.json() if decision == old_decision: log.debug('Skipped emitting fedmsg, decision did not change: %s', decision) - else: - msg = decision - decision.update({ - 'subject_type': subject_type, - 'subject_identifier': subject_identifier, - # subject is for backwards compatibility only: - 'subject': subject_type_identifier_to_list(subject_type, - subject_identifier), - 'testcase': testcase, - 'decision_context': decision_context, - 'product_version': product_version, - 'previous': old_decision, - }) - log.info( - 'Emitted a message on the bus, %r, with the topic ' - '"greenwave.decision.update"', decision) - if self.flask_app.config['MESSAGING'] == 'fedmsg': - log.debug(' - to fedmsg') + messaging_tx_stopped_counter.labels(handler='waiverdb').inc() + continue + + msg = decision + decision.update({ + 'subject_type': subject_type, + 'subject_identifier': subject_identifier, + # subject is for backwards compatibility only: + 'subject': subject_type_identifier_to_list(subject_type, + subject_identifier), + 'testcase': testcase, + 'decision_context': decision_context, + 'product_version': product_version, + 'previous': old_decision, + }) + log.info( + 'Emitting a message on the bus, %r, with the topic ' + '"greenwave.decision.update"', decision) + if self.flask_app.config['MESSAGING'] == 'fedmsg': + log.debug(' - to fedmsg') + try: fedmsg.publish(topic='decision.update', msg=msg) - elif self.flask_app.config['MESSAGING'] == 'fedora-message': - log.debug(' - to fedora-messaging') - try: - msg = fedora_messaging.api.Message( - topic='greenwave.decision.update', - body=msg - ) - fedora_messaging.api.publish(msg) - except fedora_messaging.exceptions.PublishReturned as e: - log.warning( - 'Fedora Messaging broker rejected message %s: %s', - msg.id, e) - except fedora_messaging.exceptions.ConnectionException as e: - log.warning('Error sending message %s: %s', msg.id, e) - except Exception: # pylint: disable=broad-except - log.exception('Error sending fedora-messaging message') + messaging_tx_sent_ok_counter.labels(handler='waiverdb').inc() + except Exception: + messaging_tx_failed_counter.labels(handler='waiverdb').inc() + raise + elif self.flask_app.config['MESSAGING'] == 'fedora-message': + log.debug(' - to fedora-messaging') + try: + msg = fedora_messaging.api.Message( + topic='greenwave.decision.update', + body=msg + ) + fedora_messaging.api.publish(msg) + messaging_tx_sent_ok_counter.labels(handler='waiverdb').inc() + except fedora_messaging.exceptions.PublishReturned as e: + log.warning( + 'Fedora Messaging broker rejected message %s: %s', + msg.id, e) + messaging_tx_stopped_counter.labels(handler='waiverdb').inc() + except fedora_messaging.exceptions.ConnectionException as e: + log.warning('Error sending message %s: %s', msg.id, e) + messaging_tx_failed_counter.labels(handler='waiverdb').inc() + except Exception: # pylint: disable=broad-except + log.exception('Error sending fedora-messaging message') + messaging_tx_failed_counter.labels(handler='waiverdb').inc() + + messaging_tx_stopped_counter.labels(handler='waiverdb').inc() diff --git a/greenwave/monitor.py b/greenwave/monitor.py new file mode 100644 index 0000000..6925898 --- /dev/null +++ b/greenwave/monitor.py @@ -0,0 +1,98 @@ +# SPDX-License-Identifier: GPL-2.0+ + +# For an up-to-date version of this module, see: +# https://pagure.io/monitor-flask-sqlalchemy + +import os +import tempfile + +from flask import Blueprint, Response +from prometheus_client import ( # noqa: F401 + ProcessCollector, CollectorRegistry, Counter, multiprocess, + Histogram, generate_latest, start_http_server, CONTENT_TYPE_LATEST) + +# Service-specific imports + + +if not os.environ.get('prometheus_multiproc_dir'): + os.environ.setdefault('prometheus_multiproc_dir', tempfile.mkdtemp()) +registry = CollectorRegistry() +ProcessCollector(registry=registry) +multiprocess.MultiProcessCollector(registry) +if os.getenv('MONITOR_STANDALONE_METRICS_SERVER_ENABLE', 'false') == 'true': + port = os.getenv('MONITOR_STANDALONE_METRICS_SERVER_PORT', '10040') + start_http_server(int(port), registry=registry) + + +# Generic metrics +messaging_rx_counter = Counter( + 'messaging_rx', + 'Total number of messages received', + labelnames=['handler'], + registry=registry) +messaging_rx_ignored_counter = Counter( + 'messaging_rx_ignored', + 'Number of received messages, which were ignored', + registry=registry) +messaging_rx_processed_ok_counter = Counter( + 'messaging_rx_processed_ok', + 'Number of received messages, which were processed successfully', + labelnames=['handler'], + registry=registry) +messaging_rx_failed_counter = Counter( + 'messaging_rx_failed', + 'Number of received messages, which failed during processing', + labelnames=['handler'], + registry=registry) + +messaging_tx_to_send_counter = Counter( + 'messaging_tx_to_send', + 'Total number of messages to send', + labelnames=['handler'], + registry=registry) +messaging_tx_stopped_counter = Counter( + 'messaging_tx_stopped', + 'Number of messages, which were eventually stopped before sending', + labelnames=['handler'], + registry=registry) +messaging_tx_sent_ok_counter = Counter( + 'messaging_tx_sent_ok', + 'Number of messages, which were sent successfully', + labelnames=['handler'], + registry=registry) +messaging_tx_failed_counter = Counter( + 'messaging_tx_failed', + 'Number of messages, for which the sender failed', + labelnames=['handler'], + registry=registry) + +# Service-specific metrics +# https://github.com/prometheus/client_python/issues/210 +# pylint: disable-msg=unexpected-keyword-arg,no-value-for-parameter +decision_exception_counter = Counter( + 'total_decision_exceptions', + 'All exceptions occurred in Greenwave "decision" API', + registry=registry) +decision_request_duration_seconds = Histogram( + 'decision_request_duration_seconds', + 'Decision latency', + registry=registry) +publish_decision_exceptions_waiver_counter = Counter( + 'publish_decision_exceptions_new_waiver', + 'All exceptions occurred in publishing a message after a new waiver', + registry=registry) +publish_decision_exceptions_result_counter = Counter( + 'publish_decision_exceptions_new_result', + 'All exceptions occurred in publishing a message after a new result', + registry=registry) + + +monitor_api = Blueprint( + 'monitor', __name__, + url_prefix='') + + +@monitor_api.route('/metrics') +def metrics(): + return Response(generate_latest(registry), + content_type=CONTENT_TYPE_LATEST) diff --git a/greenwave/monitoring.py b/greenwave/monitoring.py deleted file mode 100644 index 42b6780..0000000 --- a/greenwave/monitoring.py +++ /dev/null @@ -1,30 +0,0 @@ -# SPDX-License-Identifier: GPL-2.0+ - -import os -from prometheus_client import Counter, Histogram, multiprocess, CollectorRegistry - - -# tmp dir for Prometheus monitoring registry. -# Putting this here and not in the "create_app" function, because "create_app" imports the api -# ...so the check for this env variable would be made before the "create_app" can be able to set it -# This is executed only once at server starting, so it is not so bad for performace. -if not os.environ.get('prometheus_multiproc_dir'): - os.environ.setdefault('prometheus_multiproc_dir', '/tmp') -registry = CollectorRegistry() -multiprocess.MultiProcessCollector(registry) -# https://github.com/prometheus/client_python/issues/210 -# pylint: disable-msg=unexpected-keyword-arg,no-value-for-parameter -decision_exception_counter = Counter('total_decision_exceptions', ('All exceptions occurred in ' - 'Greenwave "decision" API'), - registry=registry) -decision_request_duration_seconds = Histogram('decision_request_duration_seconds', - 'Decision latency', - registry=registry) -publish_decision_exceptions_waiver_counter = Counter('publish_decision_exceptions_new_waiver', - ('All exceptions occurred in publishing a ' - 'message after a new waiver'), - registry=registry) -publish_decision_exceptions_result_counter = Counter('publish_decision_exceptions_new_result', - ('All exceptions occurred in publishing a ' - 'message after a new result'), - registry=registry)