fixed definition order broken by auto-refac; reduced prometheus code to only forwarding to analysis endpoint

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

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)

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

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