Generic error catching in DB

backend/Pipfile

@@ -27,6 +27,7 @@ rocrate = "*"
 [dev-packages]
 requests = "*"
 pytest = "*"
+pytest-asyncio = "*"
 black = "*"
 faker = "*"
 

backend/app/database/errors.py

@@ -0,0 +1,42 @@
+import logging
+import traceback
+import motor.motor_asyncio
+from typing import Optional, Generator
+from fastapi import Depends
+from pymongo import MongoClient
+
+from app.config import settings
+from app.mongo import crete_mongo_indexes
+from app.models.errors import Error
+from app.models.metadata import MongoDBRef
+
+logger = logging.getLogger(__name__)
+
+
+async def _get_db() -> Generator:
+    """Duplicate of app.dependencies.get_db(), but importing that causes circular import."""
+    mongo_client = motor.motor_asyncio.AsyncIOMotorClient(settings.MONGODB_URL)
+    db = mongo_client[settings.MONGO_DATABASE]
+    await crete_mongo_indexes(db)
+    yield db
+
+
+async def log_error(
+    exception: Exception,
+    resource: Optional[MongoDBRef] = None,
+    user: Optional[str] = None,
+):
+    """Insert new Error into the database.
+
+    Arguments:
+        exception -- instance of an Exception or subclass
+        resource -- if error relates to a specific resource, you can include it
+        user --- if error relates to actions performed by a user, you can include them
+    """
+    db = _get_db()
+    message = str(exception)
+    trace = traceback.format_exc(exception, limit=4)
+
+    logger.error(message)
+    error_log = Error(message=message, trace=trace, resource=resource, user=user)
+    await db["errors"].insert_one(error_log.to_mongo())

backend/app/main.py

@@ -145,7 +145,7 @@
 @app.on_event("startup")
 async def startup_elasticsearch():
     # create elasticsearch indices
-    es = connect_elasticsearch()
+    es = await connect_elasticsearch()
     create_index(
         es, "file", settings.elasticsearch_setting, indexSettings.file_mappings
     )

backend/app/models/errors.py

@@ -0,0 +1,25 @@
+from datetime import datetime
+from typing import Optional
+from pydantic import Field
+
+from app.models.mongomodel import MongoModel
+from app.models.metadata import MongoDBRef
+
+
+class ServiceUnreachable(Exception):
+    """Raised when Clowder can't connect to an outside service e.g. MongoDB, Elasticsearch."""
+
+    def __init__(self, service, *args):
+        super().__init__(args)
+        self.service = service
+
+    def __str__(self):
+        return f"{self.service} could not be reached."
+
+
+class Error(MongoModel):
+    message: str  # Shorthand message of the error
+    trace: str  # Full stack trace of the error
+    resource: Optional[MongoDBRef] = None
+    user_id: Optional[str] = None
+    timestamp: datetime = Field(default_factory=datetime.utcnow)

backend/app/routers/datasets.py

@@ -31,6 +31,7 @@
 from app import dependencies
 from app import keycloak_auth
 from app.search.connect import (
+    connect_elasticsearch,
     insert_record,
     delete_document_by_id,
     update_record,
@@ -191,6 +192,7 @@ async def save_dataset(
     db: MongoClient = Depends(dependencies.get_db),
     es=Depends(dependencies.get_elasticsearchclient),
 ):
+    es = await connect_elasticsearch()
 
     # Check all connection and abort if any one of them is not available
     if db is None or es is None:
@@ -304,6 +306,7 @@ async def edit_dataset(
     user_id=Depends(get_user),
     es=Depends(dependencies.get_elasticsearchclient),
 ):
+    es = await connect_elasticsearch()
 
     # Check all connection and abort if any one of them is not available
     if db is None or es is None:
@@ -347,6 +350,7 @@ async def patch_dataset(
     db: MongoClient = Depends(dependencies.get_db),
     es=Depends(dependencies.get_elasticsearchclient),
 ):
+    es = await connect_elasticsearch()
 
     # Check all connection and abort if any one of them is not available
     if db is None or es is None:
@@ -388,6 +392,7 @@ async def delete_dataset(
     fs: Minio = Depends(dependencies.get_fs),
     es=Depends(dependencies.get_elasticsearchclient),
 ):
+    es = await connect_elasticsearch()
 
     # Check all connection and abort if any one of them is not available
     if db is None or fs is None or es is None:

backend/app/routers/elasticsearch.py

@@ -7,26 +7,26 @@
 
 @router.put("/search", response_model=str)
 async def search(index_name: str, query: str):
-    es = connect_elasticsearch()
+    es = await connect_elasticsearch()
     return search_index(es, index_name, query)
 
 
 @router.post("/file/_msearch")
 async def search_file(request: Request):
-    es = connect_elasticsearch()
+    es = await connect_elasticsearch()
     query = await request.body()
     return search_index(es, "file", query)
 
 
 @router.post("/dataset/_msearch")
 async def search_dataset(request: Request):
-    es = connect_elasticsearch()
+    es = await connect_elasticsearch()
     query = await request.body()
     return search_index(es, "dataset", query)
 
 
 @router.post("/file,dataset/_msearch")
 async def search_file_and_dataset(request: Request):
-    es = connect_elasticsearch()
+    es = await connect_elasticsearch()
     query = await request.body()
     return search_index(es, ["file", "dataset"], query)

backend/app/routers/files.py

@@ -22,6 +22,7 @@
 from app import dependencies
 from app.config import settings
 from app.search.connect import (
+    connect_elasticsearch,
     insert_record,
     delete_document_by_id,
     update_record,
@@ -55,6 +56,7 @@ async def add_file_entry(
         file_db: FileDB object controlling dataset and folder destination
         file: bytes to upload
     """
+    es = await connect_elasticsearch()
 
     # Check all connection and abort if any one of them is not available
     if db is None or fs is None or es is None:
@@ -124,6 +126,8 @@ async def remove_file_entry(
     """Remove FileDB object into MongoDB, Minio, and associated metadata and version information."""
     # TODO: Deleting individual versions will require updating version_id in mongo, or deleting entire document
 
+    es = await connect_elasticsearch()
+
     # Check all connection and abort if any one of them is not available
     if db is None or fs is None or es is None:
         raise HTTPException(status_code=503, detail="Service not available")
@@ -146,6 +150,7 @@ async def update_file(
     file: UploadFile = File(...),
     es=Depends(dependencies.get_elasticsearchclient),
 ):
+    es = await connect_elasticsearch()
 
     # Check all connection and abort if any one of them is not available
     if db is None or fs is None or es is None:

backend/app/search/connect.py

@@ -7,21 +7,26 @@
 from app.models.files import FileOut
 from app.models.search import SearchCriteria
 from app.models.feeds import SearchObject
+from app.models.errors import ServiceUnreachable
+from app.database.errors import log_error
 
 logger = logging.getLogger(__name__)
 no_of_shards = settings.elasticsearch_no_of_shards
 no_of_replicas = settings.elasticsearch_no_of_replicas
 
 
-def connect_elasticsearch():
+async def connect_elasticsearch():
     """To connect to elasticsearch server and return the elasticsearch client"""
     _es = None
     logger.info(settings.elasticsearch_url)
     _es = Elasticsearch(settings.elasticsearch_url)
-    if _es.ping():
-        logger.info("Successfully connected to Elasticsearch")
-    else:
-        logger.info("Can not connect to Elasticsearch")
+    try:
+        if _es.ping():
+            logger.info("Successfully connected to Elasticsearch")
+        else:
+            raise ServiceUnreachable("Elasticsearch")
+    except ServiceUnreachable as e:
+        await log_error(e)
     return _es
 
 

backend/app/tests/test_elastic_search.py

@@ -1,5 +1,6 @@
 import json
 import time
+import pytest
 from datetime import datetime
 
 from bson import ObjectId
@@ -57,8 +58,9 @@
 }
 
 
-def test_files():
-    es = connect_elasticsearch()
+@pytest.mark.asyncio
+async def test_files():
+    es = await connect_elasticsearch()
     if es is not None:
         create_index(
             es,
@@ -95,8 +97,9 @@ def test_files():
         delete_index(es, dummy_file_index_name)
 
 
-def test_datasets():
-    es = connect_elasticsearch()
+@pytest.mark.asyncio
+async def test_datasets():
+    es = await connect_elasticsearch()
     if es is not None:
         create_index(
             es,