Bump dockerflow from 2022.8.0 to 2024.4.2 (#932)

* Bump dockerflow from 2022.8.0 to 2024.4.2

Bumps [dockerflow](https://github.com/mozilla-services/python-dockerflow) from 2022.8.0 to 2024.4.2.
- [Release notes](https://github.com/mozilla-services/python-dockerflow/releases)
- [Changelog](https://github.com/mozilla-services/python-dockerflow/blob/main/docs/changelog.rst)
- [Commits](mozilla-services/python-dockerflow@2022.8.0...2024.04.2)

---
updated-dependencies:
- dependency-name: dockerflow
  dependency-type: direct:production
  update-type: version-update:semver-major
...

Signed-off-by: dependabot[bot] <[email protected]>

* Move test to test folder

* Leverage version and lbheartbeat endpoints

* Leverage checks system for heartbeat endpoint

* Make lint happy

* Leverage request id context

* Restore local change for DB port

* Do not use FastAPI dependency for docker check

* Move fixtures into tests where used once

---------

Signed-off-by: dependabot[bot] <[email protected]>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
Co-authored-by: Mathieu Leplatre <[email protected]>

ctms/app.py

@@ -1,9 +1,10 @@
 import sys
 import time
-from secrets import token_hex
 
 import structlog
 import uvicorn
+from dockerflow.fastapi import router as dockerflow_router
+from dockerflow.fastapi.middleware import RequestIdMiddleware
 from fastapi import FastAPI, Request
 from sentry_sdk.integrations.asgi import SentryAsgiMiddleware
 
@@ -26,6 +27,7 @@
     description="CTMS API (work in progress)",
     version=get_version()["version"],
 )
+app.include_router(dockerflow_router)
 app.include_router(platform.router)
 app.include_router(contacts.router)
 
@@ -74,13 +76,7 @@ async def log_request_middleware(request: Request, call_next):
     return response
 
 
-@app.middleware("http")
-async def request_id(request: Request, call_next):
-    """Read the request id from headers. This is set by NGinx."""
-    request.state.rid = request.headers.get("X-Request-Id", token_hex(16))
-    response = await call_next(request)
-    return response
-
+app.add_middleware(RequestIdMiddleware)
 
 if __name__ == "__main__":
     uvicorn.run("app:app", host="0.0.0.0", port=80, reload=True)

ctms/log.py

@@ -6,6 +6,7 @@
 from typing import Any, Dict, List, Optional
 
 import structlog
+from dockerflow.logging import request_id_context
 from fastapi import Request
 from starlette.routing import Match
 from structlog.types import Processor
@@ -35,6 +36,11 @@ def configure_logging(
     logging_config = {
         "version": 1,
         "disable_existing_loggers": False,
+        "filters": {
+            "request_id": {
+                "()": "dockerflow.logging.RequestIdLogFilter",
+            },
+        },
         "formatters": {
             "dev_console": {
                 "()": structlog.stdlib.ProcessorFormatter,
@@ -52,6 +58,7 @@ def configure_logging(
                 "stream": sys.stdout,
                 "formatter": "dev_console",
                 "level": logging.DEBUG,
+                "filters": ["request_id"],
             },
             "null": {
                 "class": "logging.NullHandler",
@@ -61,6 +68,7 @@ def configure_logging(
                 "stream": sys.stdout,
                 "formatter": "mozlog_json",
                 "level": logging.DEBUG,
+                "filters": ["request_id"],
             },
         },
         "root": {
@@ -111,7 +119,7 @@ def context_from_request(request: Request) -> Dict:
         "client_host": host,
         "method": request.method,
         "path": request.url.path,
-        "rid": request.state.rid,
+        "rid": request_id_context.get(),
     }
 
     # Determine the path template, like "/ctms/{email_id}"

ctms/routers/platform.py

@@ -1,27 +1,21 @@
 import logging
-import time
-from typing import Any, Optional
+from typing import Optional
 
+from dockerflow import checks as dockerflow_checks
 from fastapi import APIRouter, Depends, HTTPException, Request, Response
-from fastapi.responses import JSONResponse, RedirectResponse
+from fastapi.responses import RedirectResponse
 from fastapi.security import HTTPBasicCredentials
 from prometheus_client import CONTENT_TYPE_LATEST, generate_latest
 from sqlalchemy.orm import Session
-from structlog.testing import capture_logs
 
 from ctms.auth import (
     OAuth2ClientCredentialsRequestForm,
     create_access_token,
     verify_password,
 )
-from ctms.config import Settings, get_version
 from ctms.crud import count_total_contacts, get_api_client_by_id, ping
-from ctms.dependencies import (
-    get_db,
-    get_enabled_api_client,
-    get_settings,
-    get_token_settings,
-)
+from ctms.database import SessionLocal
+from ctms.dependencies import get_db, get_enabled_api_client, get_token_settings
 from ctms.metrics import get_metrics, get_metrics_registry, token_scheme
 from ctms.schemas.api_client import ApiClientSchema
 from ctms.schemas.web import BadRequestResponse, TokenResponse
@@ -97,50 +91,31 @@ def login(
     }
 
 
-@router.get("/__version__", tags=["Platform"])
-def version():
-    """Return version.json, as required by Dockerflow."""
-    return get_version()
-
-
-@router.get("/__heartbeat__", tags=["Platform"])
-@router.head("/__heartbeat__", tags=["Platform"])
-def heartbeat(
-    request: Request,
-    db: Session = Depends(get_db),
-    settings: Settings = Depends(get_settings),
-):
-    """Return status of backing services, as required by Dockerflow."""
+@dockerflow_checks.register
+def database():
+    result = []
 
-    result: dict[str, Any] = {}
+    with SessionLocal() as db:
+        alive = ping(db)
+        if not alive:
+            result.append(
+                dockerflow_checks.Error("Database not reachable", id="db.0001")
+            )
+            return result
+        # Report number of contacts in the database.
+        # Sending the metric in this heartbeat endpoint is simpler than reporting
+        # it in every write endpoint. Plus, performance does not matter much here
+        total_contacts = count_total_contacts(db)
 
-    start_time = time.monotonic()
-    alive = ping(db)
-    result["database"] = {
-        "up": alive,
-        "time_ms": int(round(1000 * time.monotonic() - start_time)),
-    }
-    if not alive:
-        return JSONResponse(content=result, status_code=503)
-
-    appmetrics = get_metrics()
-    # Report number of contacts in the database.
-    # Sending the metric in this heartbeat endpoint is simpler than reporting
-    # it in every write endpoint. Plus, performance does not matter much here
-    total_contacts = count_total_contacts(db)
     contact_query_successful = total_contacts >= 0
-    if appmetrics and contact_query_successful:
-        appmetrics["contacts"].set(total_contacts)
-
-    status_code = 200 if contact_query_successful else 503
-    return JSONResponse(content=result, status_code=status_code)
-
+    if contact_query_successful:
+        appmetrics = get_metrics()
+        if appmetrics:
+            appmetrics["contacts"].set(total_contacts)
+    else:
+        result.append(dockerflow_checks.Error("Contacts table empty", id="db.0002"))
 
-@router.get("/__lbheartbeat__", tags=["Platform"])
-@router.head("/__lbheartbeat__", tags=["Platform"])
-def lbheartbeat():
-    """Return response when application is running, as required by Dockerflow."""
-    return {"status": "OK"}
+    return result
 
 
 @router.get("/__crash__", tags=["Platform"], include_in_schema=False)
@@ -155,11 +130,3 @@ def metrics():
     headers = {"Content-Type": CONTENT_TYPE_LATEST}
     registry = get_metrics_registry()
     return Response(generate_latest(registry), status_code=200, headers=headers)
-
-
-def test_get_metrics(anon_client, setup_metrics):
-    """An anonoymous user can request metrics."""
-    with capture_logs() as cap_logs:
-        resp = anon_client.get("/metrics")
-    assert resp.status_code == 200
-    assert len(cap_logs) == 1

pyproject.toml

@@ -23,7 +23,7 @@ python-multipart = "^0.0.9"
 python-jose = {extras = ["cryptography"], version = "^3.2.0"}
 passlib = {extras = ["argon2"], version = "^1.7.4"}
 python-dateutil = "^2.9.0"
-dockerflow = "^2022.8.0"
+dockerflow = {extras = ["fastapi"], version = "^2024.4.2"}
 sentry-sdk = "^2.11.0"
 lxml = "^5.2.2"
 prometheus-client = "0.20.0"

tests/unit/routers/test_platform.py

@@ -1,26 +1,12 @@
 import json
 from pathlib import Path
-from unittest.mock import Mock
+from unittest import mock
 
 import pytest
 from sqlalchemy.exc import TimeoutError as SQATimeoutError
 from structlog.testing import capture_logs
 
 from ctms.app import app
-from ctms.dependencies import get_db
-
-
-@pytest.fixture
-def mock_db():
-    """Mock the database session."""
-    mocked_db = Mock()
-
-    def mock_get_db():
-        yield mocked_db
-
-    app.dependency_overrides[get_db] = mock_get_db
-    yield mocked_db
-    del app.dependency_overrides[get_db]
 
 
 def test_read_root(anon_client):
@@ -37,15 +23,21 @@ def test_read_root(anon_client):
 
 def test_read_version(anon_client):
     """__version__ returns the contents of version.json."""
+    before = getattr(app.state, "APP_DIR", None)
     here = Path(__file__)
     root_dir = here.parents[3]
+    app.state.APP_DIR = root_dir
     version_path = Path(root_dir / "version.json")
     with open(version_path, "r", encoding="utf8") as vp_file:
         version_contents = vp_file.read()
     expected = json.loads(version_contents)
+
     resp = anon_client.get("/__version__")
+
     assert resp.status_code == 200
     assert resp.json() == expected
+    if before is not None:
+        app.state.APP_DIR = before
 
 
 def test_crash_authorized(client):
@@ -67,34 +59,49 @@ def test_read_heartbeat(anon_client):
         resp = anon_client.get("/__heartbeat__")
     assert resp.status_code == 200
     data = resp.json()
-    expected = {"database": {"up": True, "time_ms": data["database"]["time_ms"]}}
-    assert data == expected
+    assert data == {
+        "checks": {"database": "ok"},
+        "details": {},
+        "status": "ok",
+    }
     assert len(cap_logs) == 1
 
 
-def test_read_heartbeat_no_db_fails(anon_client, mock_db):
+@mock.patch("ctms.routers.platform.SessionLocal")
+def test_read_heartbeat_db_fails(mock_db, anon_client):
     """/__heartbeat__ returns 503 when the database is unavailable."""
-    mock_db.execute.side_effect = SQATimeoutError()
+    mocked_session = mock.MagicMock()
+    mocked_session.__enter__.return_value = mocked_session
+    mocked_session.execute.side_effect = SQATimeoutError()
+    mock_db.return_value = mocked_session
+
     resp = anon_client.get("/__heartbeat__")
-    assert resp.status_code == 503
+    assert resp.status_code == 500
     data = resp.json()
-    expected = {"database": {"up": False, "time_ms": data["database"]["time_ms"]}}
-    assert data == expected
+    assert data == {
+        "checks": {"database": "error"},
+        "details": {
+            "database": {
+                "level": 40,
+                "messages": {
+                    "db.0001": "Database not reachable",
+                },
+                "status": "error",
+            },
+        },
+        "status": "error",
+    }
 
 
 def test_read_health(anon_client):
     """The platform calls /__lbheartbeat__ to see when the app is running."""
-    with capture_logs() as cap_logs:
-        resp = anon_client.get("/__lbheartbeat__")
+    resp = anon_client.get("/__lbheartbeat__")
     assert resp.status_code == 200
-    assert resp.json() == {"status": "OK"}
-    assert len(cap_logs) == 1
 
 
-@pytest.mark.parametrize("path", ("/__lbheartbeat__", "/__heartbeat__"))
-def test_head_monitoring_endpoints(anon_client, path):
-    """Monitoring endpoints can be called without credentials"""
+def test_get_metrics(anon_client, setup_metrics):
+    """An anonoymous user can request metrics."""
     with capture_logs() as cap_logs:
-        resp = anon_client.head(path)
+        resp = anon_client.get("/metrics")
     assert resp.status_code == 200
     assert len(cap_logs) == 1