From 5a382f22c390d380b23e470c56c900356e01849e Mon Sep 17 00:00:00 2001 From: Julius Unverfehrt Date: Thu, 17 Mar 2022 12:52:42 +0100 Subject: [PATCH 1/4] Pull request #22: Change queue durability Merge in RR/pyinfra from change-queue-durability to master Squashed commit of the following: commit b01e5e05e27776cbb629ba3c2e757a64934ace30 Author: Julius Unverfehrt Date: Thu Mar 17 12:50:40 2022 +0100 refactor commit 549b89952e626ef7b1029c97ec7fb527112cc2b2 Author: Julius Unverfehrt Date: Thu Mar 17 12:47:13 2022 +0100 black commit 6a36f56a3e6308f4954a28c1bf4797a838986371 Author: Julius Unverfehrt Date: Thu Mar 17 12:46:51 2022 +0100 adjust mock scriptis,add different file format option for IO commit 3b50f88482d74ea39240cdd41cc7dd62076c87b7 Author: Julius Unverfehrt Date: Thu Mar 17 12:45:49 2022 +0100 change request_queue to durable=True --- .../queue/queue_manager/pika_queue_manager.py | 4 +- scripts/manage_minio.py | 7 +++- scripts/mock_client.py | 40 +++++++++++++------ 3 files changed, 34 insertions(+), 17 deletions(-) diff --git a/pyinfra/queue/queue_manager/pika_queue_manager.py b/pyinfra/queue/queue_manager/pika_queue_manager.py index 6cfae52..75e1596 100644 --- a/pyinfra/queue/queue_manager/pika_queue_manager.py +++ b/pyinfra/queue/queue_manager/pika_queue_manager.py @@ -74,8 +74,8 @@ class PikaQueueManager(QueueManager): args = {"x-dead-letter-exchange": "", "x-dead-letter-routing-key": dead_letter_queue} - self.channel.queue_declare(input_queue, arguments=args, auto_delete=False) - self.channel.queue_declare(output_queue, arguments=args, auto_delete=False) + self.channel.queue_declare(input_queue, arguments=args, auto_delete=False, durable=True) + self.channel.queue_declare(output_queue, arguments=args, auto_delete=False, durable=True) def republish(self, body, n_current_attempts, frame): self.channel.basic_publish( diff --git a/scripts/manage_minio.py b/scripts/manage_minio.py index 3a72632..eaa921e 100644 --- a/scripts/manage_minio.py +++ b/scripts/manage_minio.py @@ -37,8 +37,11 @@ def upload_compressed_response(storage, bucket_name, dossier_id, file_id, result 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") + if Path(path).suffix == ".pdf": + suffix_gz = ".ORIGIN.pdf.gz" + if Path(path).suffix == ".json": + suffix_gz = ".TEXT.json.gz" + path_gz = combine_dossier_id_and_file_id_and_extension(dossier_id, Path(path).stem, suffix_gz) with open(path, "rb") as f: data = gzip.compress(f.read()) diff --git a/scripts/mock_client.py b/scripts/mock_client.py index 3590d3e..d13b0c9 100644 --- a/scripts/mock_client.py +++ b/scripts/mock_client.py @@ -1,3 +1,4 @@ +import argparse import json import pika @@ -6,6 +7,13 @@ from pyinfra.config import CONFIG from pyinfra.storage.storages import get_s3_storage +def parse_args(): + parser = argparse.ArgumentParser() + parser.add_argument("--analysis_container", "-a", choices=["detr", "ner", "image"], required=True) + args = parser.parse_args() + return args + + def read_connection_params(): credentials = pika.PlainCredentials(CONFIG.rabbitmq.user, CONFIG.rabbitmq.password) parameters = pika.ConnectionParameters( @@ -25,7 +33,7 @@ def make_channel(connection) -> pika.adapters.blocking_connection.BlockingChanne 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) + return channel.queue_declare(queue=queue, auto_delete=False, durable=True, arguments=args) def make_connection() -> pika.BlockingConnection: @@ -34,32 +42,38 @@ def make_connection() -> pika.BlockingConnection: return connection -def build_message_bodies(): +def build_message_bodies(analyse_container_type): + def update_message(message_dict): + if analyse_container_type == "detr" or analyse_container_type == "image": + message_dict.update({"targetFileExtension": "ORIGIN.pdf.gz", "responseFileExtension": "IMAGE_INFO.json.gz"}) + if analyse_container_type == "ner": + message_dict.update( + {"targetFileExtension": "TEXT.json.gz", "responseFileExtension": "NER_ENTITIES.json.gz"}) + return message_dict + storage = get_s3_storage() for bucket_name, pdf_name in storage.get_all_object_names(CONFIG.storage.bucket): file_id = pdf_name.split(".")[0] dossier_id, file_id = file_id.split("/") - yield json.dumps( - { - "dossierId": dossier_id, - "fileId": file_id, - "targetFileExtension": "ORIGIN.pdf.gz", - "responseFileExtension": "detr.json.gz", - } - ).encode() + message_dict = {"dossierId": dossier_id, "fileId": file_id} + update_message(message_dict) + yield json.dumps(message_dict).encode() -if __name__ == "__main__": - +def main(args): connection = make_connection() channel = make_channel(connection) declare_queue(channel, CONFIG.rabbitmq.queues.input) declare_queue(channel, CONFIG.rabbitmq.queues.output) - for body in build_message_bodies(): + for body in build_message_bodies(args.analysis_container): channel.basic_publish("", CONFIG.rabbitmq.queues.input, body) print(f"Put {body} on {CONFIG.rabbitmq.queues.input}") 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) + + +if __name__ == "__main__": + main(parse_args()) From ea49b001becd7c940f721732d8c78f040880100d Mon Sep 17 00:00:00 2001 From: Julius Unverfehrt Date: Thu, 17 Mar 2022 14:06:48 +0100 Subject: [PATCH 2/4] warning about missing prometheus endpoint happens only once --- pyinfra/flask.py | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/pyinfra/flask.py b/pyinfra/flask.py index 35e3d3a..8196c07 100644 --- a/pyinfra/flask.py +++ b/pyinfra/flask.py @@ -6,7 +6,6 @@ from waitress import serve from pyinfra.config import CONFIG - logger = logging.getLogger(__file__) logger.setLevel(CONFIG.service.logging_level) @@ -31,6 +30,7 @@ def run_probing_webserver(app, host=None, port=None, mode=None): def set_up_probing_webserver(): # TODO: implement meaningful checks app = Flask(__name__) + informed_about_missing_prometheus_endpoint = False @app.route("/ready", methods=["GET"]) def ready(): @@ -46,12 +46,15 @@ def set_up_probing_webserver(): @app.route("/prometheus", methods=["GET"]) def get_metrics_from_analysis_endpoint(): + nonlocal informed_about_missing_prometheus_endpoint try: resp = requests.get(f"{CONFIG.rabbitmq.callback.analysis_endpoint}/prometheus") resp.raise_for_status() return resp.text except Exception as err: - logger.warning(f"Got no metrics from analysis prometheus endpoint: {err}") + if not informed_about_missing_prometheus_endpoint: + logger.warning(f"Got no metrics from analysis prometheus endpoint: {err}") + informed_about_missing_prometheus_endpoint = True return resp return app From 1b1da50faf3c5a0a9c094c5bd37fc5769806d63d Mon Sep 17 00:00:00 2001 From: Julius Unverfehrt Date: Mon, 21 Mar 2022 15:22:38 +0100 Subject: [PATCH 3/4] Pull request #24: Response dict & Prometheus forwarding Merge in RR/pyinfra from response-dict to master Squashed commit of the following: commit aed09653aafe7b9558c542ea0c36a5f072c6ab9e Author: Julius Unverfehrt Date: Mon Mar 21 15:20:24 2022 +0100 BLACK commit d1b76673a22373c32815466c37dcb006fa0d6fb4 Author: Julius Unverfehrt Date: Mon Mar 21 15:19:55 2022 +0100 change default response dict commit 82bd48a7545b9b244bdcdebf7ce976947eecf580 Author: Julius Unverfehrt Date: Mon Mar 21 15:19:15 2022 +0100 fix handling of missing prometheus endpoint --- pyinfra/flask.py | 16 ++++++++++------ scripts/mock_client.py | 3 ++- src/serve.py | 2 ++ 3 files changed, 14 insertions(+), 7 deletions(-) diff --git a/pyinfra/flask.py b/pyinfra/flask.py index 8196c07..fcddb97 100644 --- a/pyinfra/flask.py +++ b/pyinfra/flask.py @@ -50,11 +50,15 @@ def set_up_probing_webserver(): try: resp = requests.get(f"{CONFIG.rabbitmq.callback.analysis_endpoint}/prometheus") resp.raise_for_status() - return resp.text - except Exception as err: - if not informed_about_missing_prometheus_endpoint: - logger.warning(f"Got no metrics from analysis prometheus endpoint: {err}") - informed_about_missing_prometheus_endpoint = True - return resp + except ConnectionError: + return "" + except requests.exceptions.HTTPError as err: + if resp.status_code == 404: + if not informed_about_missing_prometheus_endpoint: + logger.warning(f"Got no metrics from analysis prometheus endpoint: {err}") + informed_about_missing_prometheus_endpoint = True + else: + raise err + return resp.text return app diff --git a/scripts/mock_client.py b/scripts/mock_client.py index d13b0c9..89b2b90 100644 --- a/scripts/mock_client.py +++ b/scripts/mock_client.py @@ -48,7 +48,8 @@ def build_message_bodies(analyse_container_type): message_dict.update({"targetFileExtension": "ORIGIN.pdf.gz", "responseFileExtension": "IMAGE_INFO.json.gz"}) if analyse_container_type == "ner": message_dict.update( - {"targetFileExtension": "TEXT.json.gz", "responseFileExtension": "NER_ENTITIES.json.gz"}) + {"targetFileExtension": "TEXT.json.gz", "responseFileExtension": "NER_ENTITIES.json.gz"} + ) return message_dict storage = get_s3_storage() diff --git a/src/serve.py b/src/serve.py index 8b76724..1948abf 100644 --- a/src/serve.py +++ b/src/serve.py @@ -31,6 +31,8 @@ def make_callback(analysis_endpoint): operations = message.get("operations", ["/"]) results = map(perform_operation, operations) result = dict(zip(operations, results)) + if list(result.keys()) == ["/"]: + result = list(result.values())[0] return result return callback From 11c9f8a073ef228768fe31605d411129e1f79a4c Mon Sep 17 00:00:00 2001 From: Julius Unverfehrt Date: Tue, 22 Mar 2022 15:48:27 +0100 Subject: [PATCH 4/4] Pull request #25: Fixes Merge in RR/pyinfra from fixes to master Squashed commit of the following: commit e3eff12ccdea52e041cc7a14cda72d3e32aa2144 Author: Julius Unverfehrt Date: Tue Mar 22 15:42:35 2022 +0100 black commit 2bc520c849ea4e833cb60b2c97626da6636d3155 Author: Julius Unverfehrt Date: Tue Mar 22 15:42:08 2022 +0100 adjust mock script commit 429b6b8f3a3fc8aa35515395712057d1c7bec13e Author: Julius Unverfehrt Date: Tue Mar 22 15:41:42 2022 +0100 change scope for retry consume commit 7488394e313270fe7ba356c40d810e7cb3c706ee Author: Julius Unverfehrt Date: Tue Mar 22 15:39:39 2022 +0100 add heartbeat to AMQP connection commit 004c5fa805bfb982f55de533bc109fa21bacfbc8 Author: Julius Unverfehrt Date: Tue Mar 22 15:38:15 2022 +0100 Adjust error handling for missing prometheus endpoint: error is logged not raised --- pyinfra/config.py | 15 +++++++-------- pyinfra/flask.py | 2 +- pyinfra/queue/queue_manager/pika_queue_manager.py | 9 +++++++-- scripts/mock_client.py | 3 +++ src/serve.py | 4 ++-- 5 files changed, 20 insertions(+), 13 deletions(-) diff --git a/pyinfra/config.py b/pyinfra/config.py index 5927171..937b760 100644 --- a/pyinfra/config.py +++ b/pyinfra/config.py @@ -7,14 +7,13 @@ from pyinfra.locations import CONFIG_FILE def make_art(): return """ -______ _____ __ -| ___ \ |_ _| / _| -| |_/ / _ | | _ __ | |_ _ __ __ _ -| __/ | | || || '_ \| _| '__/ _` | -| | | |_| || || | | | | | | | (_| | -\_| \__, \___/_| |_|_| |_| \__,_| - __/ | - |___/ + ___ _ _ ___ __ + o O O | _ \ | || | |_ _| _ _ / _| _ _ __ _ + o | _/ \_, | | | | ' \ | _| | '_| / _` | + TS__[O] _|_|_ _|__/ |___| |_||_| _|_|_ _|_|_ \__,_| + {======|_| ``` |_| ````|_|`````|_|`````|_|`````|_|`````|_|`````| +./o--000' `-0-0-' `-0-0-' `-0-0-' `-0-0-' `-0-0-' `-0-0-' `-0-0-' + """ diff --git a/pyinfra/flask.py b/pyinfra/flask.py index fcddb97..549ef0c 100644 --- a/pyinfra/flask.py +++ b/pyinfra/flask.py @@ -58,7 +58,7 @@ def set_up_probing_webserver(): logger.warning(f"Got no metrics from analysis prometheus endpoint: {err}") informed_about_missing_prometheus_endpoint = True else: - raise err + logging.warning(f"Caught {err}") return resp.text return app diff --git a/pyinfra/queue/queue_manager/pika_queue_manager.py b/pyinfra/queue/queue_manager/pika_queue_manager.py index 75e1596..3b91fea 100644 --- a/pyinfra/queue/queue_manager/pika_queue_manager.py +++ b/pyinfra/queue/queue_manager/pika_queue_manager.py @@ -45,7 +45,12 @@ def get_connection(): credentials = pika.PlainCredentials(username=CONFIG.rabbitmq.user, password=CONFIG.rabbitmq.password) - kwargs = {"host": CONFIG.rabbitmq.host, "port": CONFIG.rabbitmq.port, "credentials": credentials} + kwargs = { + "host": CONFIG.rabbitmq.host, + "port": CONFIG.rabbitmq.port, + "credentials": credentials, + "heartbeat": CONFIG.rabbitmq.heartbeat, + } parameters = pika.ConnectionParameters(**kwargs) @@ -96,7 +101,7 @@ class PikaQueueManager(QueueManager): def publish_response(self, message, callback, max_attempts=3): - logger.debug(f"Publishing response for {message}.") + logger.debug(f"Processing {message}.") frame, properties, body = message diff --git a/scripts/mock_client.py b/scripts/mock_client.py index 89b2b90..e7699fa 100644 --- a/scripts/mock_client.py +++ b/scripts/mock_client.py @@ -54,6 +54,8 @@ def build_message_bodies(analyse_container_type): storage = get_s3_storage() for bucket_name, pdf_name in storage.get_all_object_names(CONFIG.storage.bucket): + if "pdf" not in pdf_name: + continue file_id = pdf_name.split(".")[0] dossier_id, file_id = file_id.split("/") message_dict = {"dossierId": dossier_id, "fileId": file_id} @@ -74,6 +76,7 @@ def main(args): 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) + channel.close() if __name__ == "__main__": diff --git a/src/serve.py b/src/serve.py index 1948abf..7e26ce4 100644 --- a/src/serve.py +++ b/src/serve.py @@ -54,11 +54,11 @@ def main(): @retry(ConsumerError, tries=3, delay=5, jitter=(1, 3)) def consume(): + consumer = Consumer(visitor, queue_manager) try: - consumer = Consumer(visitor, queue_manager) consumer.consume_and_publish() except Exception as err: - raise ConsumerError from err + raise ConsumerError() from err try: consume()