skip: Refactor/remove mq (merge commit)

Merge branch 'refactor/remove-mq' into 'main' * remove mq from architecture See merge request https://gitlab.ci.csc.fi/sds-dev/sd-submit/metadata-submitter/-/merge_requests/850 Approved-by: Felipe Morato Moreira <felipmor@csc.fi> Co-authored-by: Stefan Negru <stefan@blankdots.com> Merged by Felipe Morato Moreira <felipmor@csc.fi>
CSCfi · Jun 6, 2024 · 7a013f2 · 7a013f2
2 parents 8ba90fc + d9ab7b2
commit 7a013f2
Show file tree

Hide file tree

Showing 28 changed files with 11 additions and 1,043 deletions.
diff --git a/.env.example b/.env.example
@@ -27,20 +27,6 @@ MONGO_SSL=True
 MONGO_SSL_CA=/tls/cacert
 MONGO_SSL_CLIENT_CERT_KEY=/tls/combined
 
-# rabbitmq
-BROKER_USERNAME=admin
-BROKER_PASSWORD=admin
-BROKER_HOST=messagebroker
-BROKER_PORT=5672
-BROKER_PORT_SSL=5671
-BROKER_SSL=False
-BROKER_CA=/tls/cacert
-BROKER_SSL_CLIENTCERT=/tls/cert
-BROKER_SSL_CLIENTKEY=/tls/key
-BROKER_MANAGEMENT_PORT=15672
-BROKER_MANAGEMENT_PORT_SSL=15671
-BROKER_CONFIG_FILE=/etc/rabbitmq/rabbitmq.conf
-
 # doi
 DOI_API=http://mockdoi:8001
 DOI_PREFIX=10.xxxx

diff --git a/Procfile b/Procfile
@@ -1,5 +1,4 @@
 db:             docker run -v ${PWD}/scripts/mongodb/init_mongo.js:/docker-entrypoint-initdb.d/init_mongo.js:ro --env-file ${PWD}/.env -p 27017:27017 -p 28017:28017 mongo
-messagebroker:  docker run -v ${PWD}/scripts/rabbitmq/rabbitmq.conf:/etc/rabbitmq/rabbitmq.conf:ro -v ${PWD}/scripts/rabbitmq/rabbitmq-definitions.json:/etc/rabbitmq/rabbitmq-definitions.json:ro -v mq:/var/lib/rabbitmq --env-file ${PWD}/.env -p 15672:15672 -p 5672:5672 rabbitmq:3.11.2-management-alpine
 mockauth:   gunicorn --reload --worker-class aiohttp.GunicornUVLoopWebWorker --workers 1 --log-level debug --graceful-timeout 60 --timeout 120 --bind ${BACKEND_HOST}:8000              tests.integration.mock_auth:init
 mockdoi:    gunicorn --reload --worker-class aiohttp.GunicornUVLoopWebWorker --workers 1 --log-level debug --graceful-timeout 60 --timeout 120 --bind ${BACKEND_HOST}:8001              tests.integration.mock_doi_api:init
 mockmetax:  gunicorn --reload --worker-class aiohttp.GunicornUVLoopWebWorker --workers 1 --log-level debug --graceful-timeout 60 --timeout 120 --bind ${BACKEND_HOST}:8002              tests.integration.mock_metax_api:init

diff --git a/deploy/consumer.sh b/deploy/consumer.sh
diff --git a/docker-compose-tls.yml b/docker-compose-tls.yml
@@ -1,17 +1,5 @@
 version: "3.4"
 services:
-  certfixer:
-    command:
-      - /bin/sh
-      - -c
-      - |
-        chown -R 100.101 /origcerts/mq/*
-        ls -la /origcerts/mq
-        ls -la /origcerts/
-    container_name: certfixer
-    image: alpine:latest
-    volumes:
-      - ./config:/origcerts
   backend:
     build:
       dockerfile: ./dockerfiles/Dockerfile-dev
@@ -28,7 +16,6 @@ services:
       - mockauth
       - mockdoi
       - mockmetax
-      - messagebroker
     restart: on-failure
     environment:
       - "MONGO_HOST=${MONGO_HOST}"
@@ -57,15 +44,6 @@ services:
       - "REMS_USER_ID=${REMS_USER_ID}"
       - "REMS_KEY=${REMS_KEY}"
       - "REMS_URL=${REMS_URL}"
-      - "BROKER_SSL=True"
-      - "BROKER_HOST=${BROKER_HOST}"
-      - "BROKER_PORT=${BROKER_PORT_SSL}"
-      - "BROKER_USERNAME=${BROKER_USERNAME}"
-      - "BROKER_PASSWORD=${BROKER_PASSWORD}"
-      - "BROKER_CA=${BROKER_CA}"
-      - "BROKER_MANAGEMENT_PORT=${BROKER_MANAGEMENT_PORT_SSL}"
-      - "BROKER_SSL_CLIENTKEY=${BROKER_SSL_CLIENTKEY}"
-      - "BROKER_SSL_CLIENTCERT=${BROKER_SSL_CLIENTCERT}"
   database:
     image: "mongo"
     container_name: "metadata_submitter_database_dev"
@@ -82,25 +60,6 @@ services:
       - 27017
     ports:
       - "27017:27017"
-  messagebroker:
-    image: "rabbitmq:3.13.1-management-alpine"
-    container_name: "metadata_submitter_rabbitmq_dev"
-    restart: on-failure
-    environment:
-      - "RABBITMQ_DEFAULT_USER=${BROKER_USERNAME}"
-      - "RABBITMQ_DEFAULT_PASS=${BROKER_PASSWORD}"
-      - RABBITMQ_CONFIG_FILE=/etc/rabbitmq/rabbitmq-tls.conf
-    expose:
-      - 5671
-      - 15671
-    ports:
-      - "15671:15671"
-      - "5671:5671"
-    volumes:
-      - mq:/var/lib/rabbitmq
-      - ./config/mq:/tls
-      - ./scripts/rabbitmq/rabbitmq-tls.conf:/etc/rabbitmq/rabbitmq-tls.conf
-      - ./scripts/rabbitmq/rabbitmq-definitions.json:/etc/rabbitmq/rabbitmq-definitions.json
   mockauth:
     build:
       dockerfile: ./dockerfiles/Dockerfile-dev
@@ -163,4 +122,3 @@ services:
     entrypoint: [ "python", "/mock_rems_api.py", "0.0.0.0", "8003" ]
 volumes:
   data:
-  mq:
diff --git a/docker-compose.yml b/docker-compose.yml
@@ -14,7 +14,6 @@ services:
       - mockauth
       - mockdoi
       - mockmetax
-      - messagebroker
     restart: on-failure
     environment:
       - "MONGO_HOST=${MONGO_HOST}"
@@ -40,11 +39,6 @@ services:
       - "REMS_USER_ID=${REMS_USER_ID}"
       - "REMS_KEY=${REMS_KEY}"
       - "REMS_URL=${REMS_URL}"
-      - "BROKER_HOST=${BROKER_HOST}"
-      - "BROKER_PORT=${BROKER_PORT}"
-      - "BROKER_MANAGEMENT_PORT=${BROKER_MANAGEMENT_PORT}"
-      - "BROKER_USERNAME=${BROKER_USERNAME}"
-      - "BROKER_PASSWORD=${BROKER_PASSWORD}"
   database:
     image: "mongo"
     container_name: "metadata_submitter_database_dev"
@@ -59,24 +53,6 @@ services:
       - 27017
     ports:
       - "27017:27017"
-  messagebroker:
-    image: "rabbitmq:3.13.1-management-alpine"
-    container_name: "metadata_submitter_rabbitmq_dev"
-    restart: on-failure
-    environment:
-      - "RABBITMQ_DEFAULT_USER=${BROKER_USERNAME}"
-      - "RABBITMQ_DEFAULT_PASS=${BROKER_PASSWORD}"
-      - RABBITMQ_CONFIG_FILE=/etc/rabbitmq/rabbitmq.conf
-    expose:
-      - 5672
-      - 15672
-    ports:
-      - "15672:15672"
-      - "5672:5672"
-    volumes:
-      - mq:/var/lib/rabbitmq
-      - ./scripts/rabbitmq/rabbitmq.conf:/etc/rabbitmq/rabbitmq.conf
-      - ./scripts/rabbitmq/rabbitmq-definitions.json:/etc/rabbitmq/rabbitmq-definitions.json
   mockauth:
     build:
       dockerfile: dockerfiles/Dockerfile-dev
@@ -139,4 +115,3 @@ services:
     entrypoint: [ "python", "/mock_rems_api.py", "0.0.0.0", "8003" ]
 volumes:
   data:
-  mq:
diff --git a/metadata_backend/api/handlers/publish.py b/metadata_backend/api/handlers/publish.py
@@ -526,23 +526,6 @@ async def publish_submission(self, req: Request) -> Response:
         datacite_study = {}
         # check first if all the files are ready, if not return HTTPBadRequest
         await file_operator.check_submission_files_ready(submission_id)
-        if "messageBroker" in workflow.endpoints:
-            # we will only publish the files which are ready
-            files = await file_operator.read_submission_files(submission_id, ["ready"])
-            for file in files:
-                ingest_msg = {
-                    "type": "ingest",
-                    "user": external_user_id,
-                    "filepath": file["path"],
-                    "encrypted_checksums": file["encrypted_checksums"],
-                }
-                self.mq_publisher.send_message(
-                    workflow.get_endpoint_conf("messageBroker", "endpoint"),
-                    "ingest",
-                    workflow.get_endpoint_conf("messageBroker", "exchange"),
-                    ingest_msg,
-                    "message_ingestion-trigger",
-                )
         if "datacite" in workflow.endpoints:
             try:
                 datacite_study = await self._publish_datacite(submission, obj_op, operator)

diff --git a/metadata_backend/api/handlers/restapi.py b/metadata_backend/api/handlers/restapi.py
@@ -3,16 +3,14 @@
 import json
 from collections.abc import AsyncIterator, Iterator
 from math import ceil
-from typing import Any, Optional
+from typing import Any
 
 import aiohttp_session
 import ujson
 from aiohttp import web
 from aiohttp.web import Request, Response
 from multidict import CIMultiDict
 
-from metadata_backend.message_broker.mq_service import MQPublisher
-
 from ...conf.conf import WORKFLOWS, schema_types
 from ...helpers.logger import LOG
 from ...helpers.schema_loader import JSONSchemaLoader, SchemaNotFoundException
@@ -276,14 +274,11 @@ def __init__(
         metax_handler: MetaxServiceHandler,
         datacite_handler: DataciteServiceHandler,
         rems_handler: RemsServiceHandler,
-        mq_publisher: Optional[MQPublisher] = None,
     ) -> None:
         """Endpoints should have access to metax and datacite services."""
         self.metax_handler = metax_handler
         self.datacite_handler = datacite_handler
         self.rems_handler = rems_handler
-        if mq_publisher:
-            self.mq_publisher = mq_publisher
 
     @staticmethod
     async def get_user_external_id(request: web.Request) -> str:

diff --git a/metadata_backend/api/health.py b/metadata_backend/api/health.py
@@ -5,13 +5,12 @@
 import ujson
 from aiohttp import web
 from aiohttp.web import Request, Response
-from amqpstorm import management
 from motor.motor_asyncio import AsyncIOMotorClient
 from pymongo.errors import ConnectionFailure
 
 from metadata_backend.api.auth import AAIServiceHandler
 
-from ..conf.conf import mq_config, url
+from ..conf.conf import url
 from ..helpers.logger import LOG
 from ..services.datacite_service_handler import DataciteServiceHandler
 from ..services.metax_service_handler import MetaxServiceHandler
@@ -49,13 +48,6 @@ async def get_health_status(self, _: Request) -> Response:
         else:
             services["database"] = {"status": "Down"}
 
-        _conn_mq = await self.try_mq_connection()
-        # Determine database load status
-        if _conn_mq:
-            services["messageBroker"] = {"status": "Ok"} if _conn_mq < 1000 else {"status": "Degraded"}
-        else:
-            services["messageBroker"] = {"status": "Down"}
-
         # Determine the status of loaded services
 
         services["datacite"] = await self.datacite_handler._healtcheck()
@@ -106,42 +98,3 @@ async def try_db_connection(self, db_client: AsyncIOMotorClient) -> None | float
         except ConnectionFailure:
             LOG.exception("Connection to db failed.")
             return None
-
-    async def try_mq_connection(self) -> None | float:
-        """Check the connection to RabbitMQ.
-
-        :returns: Connection time or None if connection fails
-        """
-        _ssl = mq_config["ssl"]
-        _http = "http"
-        _ssl_client_cert = None
-        if _ssl:
-            _http = "https"
-            cacertfile = str(mq_config["cacertfile"])
-            certfile = str(mq_config["certfile"])
-            keyfile = str(mq_config["keyfile"])
-            _ssl_client_cert = (certfile, keyfile)
-
-        API = management.ManagementApi(
-            f"{_http}://{mq_config['hostname']}:{mq_config['managementPort']}",
-            mq_config["username"],
-            mq_config["password"],
-            verify=cacertfile if _ssl else False,
-            cert=_ssl_client_cert,
-        )
-        try:
-            start = time.time()
-            # the ``local`` vhost should always exist in the rabbitMQ
-            result = API.aliveness_test("local")
-            if result["status"] == "ok":
-                LOG.debug("Connection to rabbitmq succeeded.")
-                perf_time = time.time() - start
-                return perf_time
-            LOG.error("Connection to rabbitmq failed, server is down.")
-            return None
-        except management.ApiConnectionError as error:
-            LOG.error("Connection to rabbitmq failed, error: %s", error)
-            return None
-        except management.ApiError as error:
-            LOG.error("Connection to rabbitmq failed due to ApiError, error: %s", error)
-            return None
diff --git a/metadata_backend/conf/conf.py b/metadata_backend/conf/conf.py
@@ -228,18 +228,3 @@ def create_db_client() -> AsyncIOMotorClient:
 
 DATACITE_SCHEMAS = {"study", "dataset", "bpdataset"}
 METAX_SCHEMAS = {"study", "dataset"}
-
-mq_ssl = str_to_bool(os.environ.get("BROKER_SSL", "False"))
-mq_port_var = "BROKER_PORT_SSL" if mq_ssl else "BROKER_PORT"
-mq_management_port_var = "BROKER_MANAGEMENT_PORT_SSL" if mq_ssl else "BROKER_MANAGEMENT_PORT"
-mq_config = {
-    "hostname": str(os.environ.get("BROKER_HOST")),
-    "port": int(os.environ.get(mq_port_var, 5671)),
-    "managementPort": int(os.environ.get(mq_management_port_var, 15671)),
-    "username": os.environ.get("BROKER_USERNAME", ""),
-    "password": os.environ.get("BROKER_PASSWORD", ""),
-    "ssl": mq_ssl,
-    "cacertfile": os.environ.get("BROKER_CA", ""),
-    "certfile": os.environ.get("BROKER_SSL_CLIENTCERT", ""),
-    "keyfile": os.environ.get("BROKER_SSL_CLIENTKEY", ""),
-}
diff --git a/metadata_backend/conf/workflows/0_fega.json b/metadata_backend/conf/workflows/0_fega.json
@@ -112,11 +112,6 @@
                 "dac",
                 "policy"
             ]
-        },
-        {
-            "name": "messageBroker",
-            "endpoint": "federated",
-            "exchange": "sda"
         }
     ]
 }
diff --git a/metadata_backend/conf/workflows/1_bigpicture.json b/metadata_backend/conf/workflows/1_bigpicture.json
@@ -104,11 +104,6 @@
                 "dac",
                 "policy"
             ]
-        },
-        {
-            "name": "messageBroker",
-            "endpoint": "bigpicture",
-            "exchange": "sda"
         }
     ]
 }
diff --git a/metadata_backend/conf/workflows/2_sdsx.json b/metadata_backend/conf/workflows/2_sdsx.json
@@ -80,11 +80,6 @@
                 "dac",
                 "policy"
             ]
-        },
-        {
-            "name": "messageBroker",
-            "endpoint": "sdsx",
-            "exchange": "sda"
         }
     ]
 }
diff --git a/metadata_backend/consumer.py b/metadata_backend/consumer.py