feat(otel): add otel to ingest API

ingest_api/runtime/requirements.txt

@@ -16,4 +16,6 @@ stac-pydantic @ git+https://github.com/ividito/stac-pydantic.git@3f4cb381c85749b
 xarray==2023.1.0
 xstac==1.1.0
 zarr==2.13.6
-boto3==1.24.59
+boto3==1.24.59
+aws_xray_sdk>=2.6.0,<3
+aws-lambda-powertools>=1.18.0

ingest_api/runtime/requirements_dev.txt

@@ -18,4 +18,6 @@ xstac==1.1.0
 zarr==2.13.6
 boto3==1.24.59
 moto[dynamodb, ssm]>=4.0.9,<5.0
-httpx
+httpx
+aws_xray_sdk>=2.6.0,<3
+aws-lambda-powertools>=1.18.0

ingest_api/runtime/src/main.py

@@ -7,13 +7,16 @@
 import src.dependencies as dependencies
 import src.schemas as schemas
 import src.services as services
+from aws_lambda_powertools.metrics import MetricUnit
 from src.collection_publisher import CollectionPublisher, ItemPublisher
 from src.doc import DESCRIPTION
+from src.monitoring import LoggerRouteHandler, logger, metrics, tracer
 
-from fastapi import Depends, FastAPI, HTTPException
+from fastapi import APIRouter, Depends, FastAPI, HTTPException
 from fastapi.exceptions import RequestValidationError
 from fastapi.responses import JSONResponse
 from fastapi.security import OAuth2PasswordRequestForm
+from starlette.requests import Request
 
 settings = (
     config.Settings()
@@ -37,8 +40,10 @@
     root_path=settings.root_path,
     openapi_url="/openapi.json",
     docs_url="/docs",
+    router=APIRouter(route_class=LoggerRouteHandler),
 )
 
+
 collection_publisher = CollectionPublisher()
 item_publisher = ItemPublisher()
 
@@ -222,7 +227,42 @@ def who_am_i(claims=Depends(auth.decode_token)):
     return claims
 
 
+# If the correlation header is used in the UI, we can analyze traces that originate from a given user or client
+@app.middleware("http")
+async def add_correlation_id(request: Request, call_next):
+    """Add correlation ids to all requests and subsequent logs/traces"""
+    # Get correlation id from X-Correlation-Id header if provided
+    corr_id = request.headers.get("x-correlation-id")
+    if not corr_id:
+        try:
+            # If empty, use request id from aws context
+            corr_id = request.scope["aws.context"].aws_request_id
+        except KeyError:
+            # If empty, use uuid
+            corr_id = "local"
+
+    # Add correlation id to logs
+    logger.set_correlation_id(corr_id)
+
+    # Add correlation id to traces
+    tracer.put_annotation(key="correlation_id", value=corr_id)
+
+    response = await tracer.capture_method(call_next)(request)
+    # Return correlation header in response
+    response.headers["X-Correlation-Id"] = corr_id
+    logger.info("Request completed")
+    return response
+
+
 # exception handling
 @app.exception_handler(RequestValidationError)
 async def validation_exception_handler(request, exc):
     return JSONResponse(str(exc), status_code=422)
+
+
+@app.exception_handler(Exception)
+async def general_exception_handler(request, err):
+    """Handle exceptions that aren't caught elsewhere"""
+    metrics.add_metric(name="UnhandledExceptions", unit=MetricUnit.Count, value=1)
+    logger.exception("Unhandled exception")
+    return JSONResponse(status_code=500, content={"detail": "Internal Server Error"})

ingest_api/runtime/src/monitoring.py

@@ -0,0 +1,42 @@
+"""Observability utils"""
+from typing import Callable
+
+from aws_lambda_powertools import Logger, Metrics, Tracer
+from aws_lambda_powertools.metrics import MetricUnit  # noqa: F401
+
+from fastapi import Request, Response
+from fastapi.routing import APIRoute
+
+logger: Logger = Logger(service="raster-api", namespace="veda-backend")
+metrics: Metrics = Metrics(service="raster-api", namespace="veda-backend")
+tracer: Tracer = Tracer()
+
+
+class LoggerRouteHandler(APIRoute):
+    """Extension of base APIRoute to add context to log statements, as well as record usage metricss"""
+
+    def get_route_handler(self) -> Callable:
+        """Overide route handler method to add logs, metrics, tracing"""
+        original_route_handler = super().get_route_handler()
+
+        async def route_handler(request: Request) -> Response:
+            # Add fastapi context to logs
+            ctx = {
+                "path": request.url.path,
+                "route": self.path,
+                "method": request.method,
+            }
+            logger.append_keys(fastapi=ctx)
+            logger.info("Received request")
+            metrics.add_metric(
+                name="/".join(
+                    str(request.url.path).split("/")[:2]
+                ),  # enough detail to capture search IDs, but not individual tile coords
+                unit=MetricUnit.Count,
+                value=1,
+            )
+            tracer.put_annotation(key="path", value=request.url.path)
+            tracer.capture_method(original_route_handler)(request)
+            return await original_route_handler(request)
+
+        return route_handler