Modernize and add OpenTelemetry (#320)

Author: tibdex

.github/workflows/deploy.yml

@@ -15,12 +15,12 @@ jobs:
     name: Deploy to AWS
     environment: deploy-to-aws
     steps:
-      - uses: actions/checkout@v4
-      - uses: aws-actions/configure-aws-credentials@v4
+      - uses: actions/checkout@v6.0.2
+      - uses: aws-actions/configure-aws-credentials@v6.0.0
         with:
           aws-region: ${{ vars.AWS_REGION }}
           role-to-assume: arn:aws:iam::${{ secrets.AWS_ACCOUNT_ID }}:role/${{ vars.AWS_DEPLOYMENT_ROLE }}
-      - uses: aws-actions/amazon-ecr-login@v2
+      - uses: aws-actions/amazon-ecr-login@v2.1.1
         id: log-into-ecr
       - name: Build, tag, and push Docker image to Amazon ECR
         id: build-tag-and-push-docker-image
@@ -32,13 +32,13 @@ jobs:
           echo "tag=$TAG" >> $GITHUB_OUTPUT
       - name: Inline variables in the task definition
         run: sed -i -e 's/AWS_ACCOUNT_ID/${{ secrets.AWS_ACCOUNT_ID }}/g' -e 's/AWS_DATABASE_URL_SECRET_NAME/${{ vars.AWS_DATABASE_URL_SECRET_NAME }}/g' -e 's/AWS_EXECUTION_ROLE/${{ vars.AWS_EXECUTION_ROLE }}/g' -e 's/AWS_REGION/${{ vars.AWS_REGION }}/g' task-definition.json
-      - uses: aws-actions/amazon-ecs-render-task-definition@v1
+      - uses: aws-actions/amazon-ecs-render-task-definition@v1.8.4
         id: render-task-definition
         with:
           container-name: atoti-session
           image: ${{ steps.build-tag-and-push-docker-image.outputs.tag }}
           task-definition: task-definition.json
-      - uses: aws-actions/amazon-ecs-deploy-task-definition@v2
+      - uses: aws-actions/amazon-ecs-deploy-task-definition@v2.6.1
         with:
           cluster: atoti-project-template
           service: atoti-project-template

.github/workflows/test.yml

@@ -8,14 +8,14 @@ jobs:
     name: Test
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v4
-      - uses: astral-sh/setup-uv@v3
+      - uses: actions/checkout@v6.0.2
+      - uses: astral-sh/setup-uv@v8.0.0
         with:
           enable-cache: true
           # Keep in sync with pyproject.toml's `tool.uv.required-version`.
-          version: "0.9.18"
+          version: "0.11.2"
       # Keep in sync with pyproject.toml's `project.requires-python`.
-      - run: uv python install 3.11
+      - run: uv python install 3.12
       - run: uv sync --locked
       - run: uv run -m skeleton
       - run: uv run ruff format --check

Dockerfile

@@ -4,7 +4,7 @@
 # Keep Python version in sync with:
 # - pyproject.toml's `project.requires-python`.
 # - the main stage below.
-FROM ghcr.io/astral-sh/uv:0.9.18-python3.11-bookworm-slim AS builder
+FROM ghcr.io/astral-sh/uv:0.11.2-python3.12-trixie-slim AS builder
 
 ENV UV_COMPILE_BYTECODE=1 UV_LINK_MODE=copy
 
@@ -21,7 +21,7 @@ RUN --mount=type=bind,source=skeleton,target=skeleton_tmp \
     rm -r app
 
 # Keep this synced with the `builder` stage above.
-FROM python:3.11-slim-bookworm
+FROM python:3.12-slim-trixie
 
 COPY --from=builder /venv app
 

README.md

@@ -7,6 +7,7 @@ On top of the `atoti` package, it comes with:
 - Dependency management with [uv](https://docs.astral.sh/uv)
 - Config management with [Pydantic](https://docs.pydantic.dev/2.6/concepts/pydantic_settings)
 - Testing with [pytest](https://docs.pytest.org)
+- Observability with [OpenTelemetry](https://opentelemetry.io/docs/languages/python)
 - Type checking with [ty](https://docs.astral.sh/ty)
 - Formatting and linting with [Ruff](https://docs.astral.sh/ruff)
 - Continuous testing with [GitHub Actions](https://github.com/features/actions)
@@ -30,6 +31,12 @@ To start the app:
 uv run python -m app
 ```
 
+To start the app in Docker with [Jaeger](https://www.jaegertracing.io) to observe traces:
+
+```bash
+docker compose up --build
+```
+
 Other useful commands can be found in [`test.yml`](.github/workflows/test.yml).
 
 ## Deployment

app/__main__.py

@@ -1,10 +1,24 @@
 from asyncio import run, to_thread
 from urllib.parse import urlparse
 
+from opentelemetry import trace
+from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
+from opentelemetry.sdk.resources import Resource
+from opentelemetry.sdk.trace import TracerProvider
+from opentelemetry.sdk.trace.export import BatchSpanProcessor
+
 from . import Config, start_app
 
 
+def _setup_opentelemetry() -> None:
+    resource = Resource.create({"service.name": "app"})
+    provider = TracerProvider(resource=resource)
+    provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter()))
+    trace.set_tracer_provider(provider)
+
+
 async def main() -> None:
+    _setup_opentelemetry()
     async with start_app(config=Config()) as session:
         port = urlparse(session.url).port or 80
         print(f"Session listening on port {port}")  # noqa: T201

app/create_and_join_tables.py

@@ -1,9 +1,11 @@
 import atoti as tt
 
+from .opentelemetry import traced
 from .skeleton import Skeleton
 from .util import column
 
 
+@traced
 def create_station_status_table(session: tt.Session, /) -> None:
     skeleton = Skeleton.tables.STATION_STATUS
     session.create_table(
@@ -20,6 +22,7 @@ def create_station_status_table(session: tt.Session, /) -> None:
     )
 
 
+@traced
 def create_station_information_table(session: tt.Session, /) -> None:
     skeleton = Skeleton.tables.STATION_INFORMATION
     session.create_table(
@@ -43,6 +46,7 @@ def create_station_information_table(session: tt.Session, /) -> None:
     )
 
 
+@traced
 def join_tables(session: tt.Session, /) -> None:
     skeleton = Skeleton.tables
     session.tables[skeleton.STATION_STATUS.name].join(
@@ -52,6 +56,7 @@ def join_tables(session: tt.Session, /) -> None:
     )
 
 
+@traced
 def create_and_join_tables(session: tt.Session, /) -> None:
     create_station_status_table(session)
     create_station_information_table(session)

app/create_cubes.py

@@ -1,9 +1,11 @@
 import atoti as tt
 
+from .opentelemetry import traced
 from .skeleton import Skeleton
 from .util import column, fact_based_hierarchy
 
 
+@traced
 def create_station_cube(session: tt.Session, /) -> None:
     skeleton = Skeleton.cubes.STATION
 
@@ -59,5 +61,6 @@ def create_station_cube(session: tt.Session, /) -> None:
         )
 
 
+@traced
 def create_cubes(session: tt.Session, /) -> None:
     create_station_cube(session)

app/load_tables.py

@@ -1,6 +1,5 @@
 from asyncio import gather
 from collections.abc import Iterable, Mapping
-from logging import getLogger
 from pathlib import Path
 from typing import Any, cast
 
@@ -10,13 +9,13 @@
 from pydantic import DirectoryPath, FilePath, HttpUrl
 
 from .config import Config
+from .opentelemetry import traced
 from .path import RESOURCES_DIRECTORY
 from .skeleton import Skeleton
 from .util import read_json, reverse_geocode
 
-_LOGGER = getLogger(__name__)
-
 
+@traced
 async def read_station_information(
     *,
     http_client: httpx.AsyncClient,
@@ -73,6 +72,7 @@ async def read_station_information(
     ).drop(columns=coordinates_column_names)
 
 
+@traced
 async def read_station_status(
     velib_data_base_path: HttpUrl | Path,
     /,
@@ -106,14 +106,14 @@ async def read_station_status(
     return pd.DataFrame(station_statuses)
 
 
+@traced
 async def load_tables(
     session: tt.Session,
     /,
     *,
     config: Config,
     http_client: httpx.AsyncClient,
 ) -> None:
-    _LOGGER.info("Loading tables.")
     if config.data_refresh_period is None:
         reverse_geocoding_path: HttpUrl | FilePath = (
             RESOURCES_DIRECTORY / "station_location.csv"
@@ -151,4 +151,3 @@ async def load_tables(
                 station_status_df
             ),
         )
-    _LOGGER.info("Tables loaded.")

app/opentelemetry.py

@@ -0,0 +1,31 @@
+from collections.abc import Callable, Coroutine
+from functools import wraps
+from inspect import iscoroutinefunction, isfunction
+from typing import Final
+
+from opentelemetry import trace
+
+TRACER: Final = trace.get_tracer(__name__)
+
+
+def traced[**P, R](func: Callable[P, R], /) -> Callable[P, R]:
+    """Thin wrapper around :meth:`opentelemetry.trace.Tracer.start_as_current_span`'s automatically using the decorated's function qualified name as the span name."""
+    assert isfunction(func)
+
+    if iscoroutinefunction(func):
+
+        @wraps(func)
+        async def wrapper(
+            *args: P.args, **kwargs: P.kwargs
+        ) -> Coroutine[object, object, R]:
+            with TRACER.start_as_current_span(func.__qualname__):
+                return await func(*args, **kwargs)
+
+        return wrapper  # ty: ignore[invalid-return-type]
+
+    @wraps(func)
+    def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
+        with TRACER.start_as_current_span(func.__qualname__):
+            return func(*args, **kwargs)
+
+    return wrapper

app/start_app.py

@@ -3,6 +3,7 @@
 
 import atoti as tt
 import httpx
+from opentelemetry.instrumentation.httpx import AsyncOpenTelemetryTransport
 
 from .config import Config
 from .load_tables import load_tables
@@ -12,8 +13,9 @@
 
 @asynccontextmanager
 async def start_app(*, config: Config) -> AsyncGenerator[tt.Session]:
+    httpx_transport = AsyncOpenTelemetryTransport(httpx.AsyncHTTPTransport())
     async with (
-        httpx.AsyncClient() as http_client,
+        httpx.AsyncClient(transport=httpx_transport) as http_client,
         start_session(config=config, http_client=http_client) as session,
         run_periodically(
             lambda: load_tables(session, config=config, http_client=http_client),