hotosm / fmtm

Field Mapping Tasking Manager - coordinated field mapping.
https://fmtm.hotosm.org/
GNU Affero General Public License v3.0
42 stars 44 forks source link

Add health / loadbalancer endpoints for Mozilla Dockerflow #1443

Open spwoodcock opened 5 months ago

spwoodcock commented 5 months ago

Is your feature request related to a problem? Please describe.

Describe the solution you'd like

Describe alternatives you've considered

I don't know any other good specs for this.

spwoodcock commented 5 months ago

Example implementation:

from pathlib import Path
from fastapi import Depends
from fastapi.responses import JSONResponse, Response
from loguru import logger as log
from sqlalchemy import text
from sqlalchemy.orm import Session

from app.__version__ import __version__
from app.db.database import get_db
from app.models.enums import HTTPStatus

@api.get("/__version__")
async def deployment_details():
    """Mozilla Dockerflow Spec: source, version, commit, and link to CI build."""
    details = {}

    version_path = Path("/opt/version.json")
    if version_path.exists():
        with open(version_path, "r") as version_file:
            details = json.load(version_file)
    commit = details.get("commit", "commit key was not found in file!")
    build = details.get("build", "build key was not found in file!")

    return JSONResponse(status_code=HTTPStatus.OK, content={
        "source" : "https://github.com/hotosm/fmtm", 
        "version": __version__,
        "commit" : commit or "/app/version.json not found",
        "build"  : build or "/app/version.json not found"
    })

@api.get("/__heartbeat__")
async def heartbeat_plus_db(db: Session = Depends(get_db)):
    """Heartbeat that checks that API and DB are both up and running."""
    try:
        db.execute(text("SELECT 1"))
        return Response(status_code=HTTPStatus.OK)
    except Exception as e:
        log.warning(e)
        log.warning("Server failed __heartbeat__ database connection check")
        return JSONResponse(
            status_code=HTTPStatus.INTERNAL_SERVER_ERROR, content={"error": str(e)}
        )

@api.get("/__lbheartbeat__")
async def simple_heartbeat():
    """Simple ping/pong API response."""
    return Response(status_code=HTTPStatus.OK)