From 894a6b5d4c7026b9a703a8b2cd70641e7ed7323b Mon Sep 17 00:00:00 2001 From: Matthias Bisping Date: Wed, 16 Mar 2022 11:16:39 +0100 Subject: [PATCH] fixed definition order broken by auto-refac; reduced prometheus code to only forwarding to analysis endpoint --- pyinfra/flask.py | 15 +++-------- scripts/manage_minio.py | 20 +++++++-------- scripts/mock_client.py | 56 ++++++++++++++++++++--------------------- src/serve.py | 13 +--------- 4 files changed, 43 insertions(+), 61 deletions(-) diff --git a/pyinfra/flask.py b/pyinfra/flask.py index 0a91bc3..e225fdb 100644 --- a/pyinfra/flask.py +++ b/pyinfra/flask.py @@ -2,9 +2,8 @@ import logging import requests from flask import Flask, jsonify -from retry import retry from waitress import serve -from prometheus_client import generate_latest, CollectorRegistry + from pyinfra.config import CONFIG @@ -42,11 +41,6 @@ def set_up_probing_webserver(): return resp @app.route("/prometheus", methods=["GET"]) - def prometheus(): - analysis_metrics = get_metrics_from_analysis_endpoint() - combined_metrics = f"{analysis_metrics}{generate_latest(PROM_REGISTER).decode()}" - return combined_metrics - def get_metrics_from_analysis_endpoint(): try: metric = requests.get(f"{CONFIG.rabbitmq.callback.analysis_endpoint}/prometheus") @@ -54,9 +48,8 @@ def set_up_probing_webserver(): return metric.text except requests.exceptions.ConnectionError as err: logging.warning(f"Got no metrics from analysis prometheus endpoint: {err}") - return "# WARNING No Analysis metrics\n" + resp = jsonify("no analysis metrics received") + resp.status_code = 504 + return resp return app - - -PROM_REGISTER = CollectorRegistry() diff --git a/scripts/manage_minio.py b/scripts/manage_minio.py index 4528a12..3a72632 100644 --- a/scripts/manage_minio.py +++ b/scripts/manage_minio.py @@ -26,6 +26,16 @@ def parse_args(): return args +def combine_dossier_id_and_file_id_and_extension(dossier_id, file_id, extension): + return f"{dossier_id}/{file_id}{extension}" + + +def upload_compressed_response(storage, bucket_name, dossier_id, file_id, result) -> None: + data = gzip.compress(result.encode()) + path_gz = combine_dossier_id_and_file_id_and_extension(dossier_id, file_id, CONFIG.service.response.extension) + storage.put_object(bucket_name, path_gz, data) + + def add_file_compressed(storage, bucket_name, dossier_id, path) -> None: path_gz = combine_dossier_id_and_file_id_and_extension(dossier_id, Path(path).stem, ".ORIGIN.pdf.gz") @@ -57,13 +67,3 @@ if __name__ == "__main__": elif args.command == "purge": storage.clear_bucket(bucket_name) - - -def combine_dossier_id_and_file_id_and_extension(dossier_id, file_id, extension): - return f"{dossier_id}/{file_id}{extension}" - - -def upload_compressed_response(storage, bucket_name, dossier_id, file_id, result) -> None: - data = gzip.compress(result.encode()) - path_gz = combine_dossier_id_and_file_id_and_extension(dossier_id, file_id, CONFIG.service.response.extension) - storage.put_object(bucket_name, path_gz, data) diff --git a/scripts/mock_client.py b/scripts/mock_client.py index 938561c..3590d3e 100644 --- a/scripts/mock_client.py +++ b/scripts/mock_client.py @@ -6,6 +6,34 @@ from pyinfra.config import CONFIG from pyinfra.storage.storages import get_s3_storage +def read_connection_params(): + credentials = pika.PlainCredentials(CONFIG.rabbitmq.user, CONFIG.rabbitmq.password) + parameters = pika.ConnectionParameters( + host=CONFIG.rabbitmq.host, + port=CONFIG.rabbitmq.port, + heartbeat=CONFIG.rabbitmq.heartbeat, + credentials=credentials, + ) + return parameters + + +def make_channel(connection) -> pika.adapters.blocking_connection.BlockingChannel: + channel = connection.channel() + channel.basic_qos(prefetch_count=1) + return channel + + +def declare_queue(channel, queue: str): + args = {"x-dead-letter-exchange": "", "x-dead-letter-routing-key": CONFIG.rabbitmq.queues.dead_letter} + return channel.queue_declare(queue=queue, auto_delete=False, arguments=args) + + +def make_connection() -> pika.BlockingConnection: + parameters = read_connection_params() + connection = pika.BlockingConnection(parameters) + return connection + + def build_message_bodies(): storage = get_s3_storage() for bucket_name, pdf_name in storage.get_all_object_names(CONFIG.storage.bucket): @@ -21,23 +49,6 @@ def build_message_bodies(): ).encode() -def make_channel(connection) -> pika.adapters.blocking_connection.BlockingChannel: - channel = connection.channel() - channel.basic_qos(prefetch_count=CONFIG.rabbitmq.prefetch_count) - return channel - - -def declare_queue(channel, queue: str): - args = {"x-dead-letter-exchange": "", "x-dead-letter-routing-key": CONFIG.rabbitmq.queues.dead_letter} - return channel.queue_declare(queue=queue, auto_delete=False, arguments=args) - - -def make_connection() -> pika.BlockingConnection: - parameters = read_connection_params() - connection = pika.BlockingConnection(parameters) - return connection - - if __name__ == "__main__": connection = make_connection() @@ -52,14 +63,3 @@ if __name__ == "__main__": for method_frame, _, body in channel.consume(queue=CONFIG.rabbitmq.queues.output): print(f"Received {json.loads(body)}") channel.basic_ack(method_frame.delivery_tag) - - -def read_connection_params(): - credentials = pika.PlainCredentials(CONFIG.rabbitmq.user, CONFIG.rabbitmq.password) - parameters = pika.ConnectionParameters( - host=CONFIG.rabbitmq.host, - port=CONFIG.rabbitmq.port, - heartbeat=CONFIG.rabbitmq.heartbeat, - credentials=credentials, - ) - return parameters diff --git a/src/serve.py b/src/serve.py index 75d8c96..8b76724 100644 --- a/src/serve.py +++ b/src/serve.py @@ -1,14 +1,12 @@ import logging -import tracemalloc from multiprocessing import Process import requests -from prometheus_client import Gauge from retry import retry from pyinfra.config import CONFIG, make_art from pyinfra.exceptions import AnalysisFailure, ConsumerError -from pyinfra.flask import run_probing_webserver, set_up_probing_webserver, PROM_REGISTER +from pyinfra.flask import run_probing_webserver, set_up_probing_webserver from pyinfra.queue.consumer import Consumer from pyinfra.queue.queue_manager.pika_queue_manager import PikaQueueManager from pyinfra.storage.storages import get_storage @@ -39,12 +37,6 @@ def make_callback(analysis_endpoint): def main(): - if CONFIG.service.monitoring_enabled: - tracemalloc.start() - memory_peak_metric = Gauge( - "pyinfra_memory_peak_per_call", "Pyinfra memory peak when processing a message in Mb" - ) - PROM_REGISTER.register(memory_peak_metric) webserver = Process(target=run_probing_webserver, args=(set_up_probing_webserver(),)) logging.info(make_art()) @@ -63,9 +55,6 @@ def main(): try: consumer = Consumer(visitor, queue_manager) consumer.consume_and_publish() - if CONFIG.service.monitoring_enabled: - _, peak = tracemalloc.get_traced_memory() - memory_peak_metric.set(peak / 10**6) except Exception as err: raise ConsumerError from err