Add observability (and logging) with request-id to APIs

Implement middleware for logging request-id(s) and other parameters in all APIs, enhancing traceability and monitoring capabilities!

auth-domain/docker-compose.dev.yml

@@ -14,6 +14,9 @@ services:
         condition: service_healthy
     networks:
       - auth
+    logging:
+      options:
+        tag: dev-auth
 
   auth-db:
     extends:

auth-domain/docker-compose.yml

@@ -11,6 +11,9 @@ services:
         condition: service_healthy
     networks:
       - auth
+    logging:
+      options:
+        tag: auth
 
   auth-db:
     extends:

auth-domain/docker-template.yml

@@ -15,6 +15,10 @@ services:
       - PORT=5000
       - DATABASE_URL=postgresql://${POSTGRES_USER}:${POSTGRES_PASS}@auth-db/${POSTGRES_DB}
       - APP_SETTINGS=${APP_SETTINGS}
+    logging:
+      driver: gelf
+      options:
+        gelf-address: "udp://localhost:12201"
 
   auth-db:
     build:

auth-domain/user-manager/requirements.txt

@@ -8,3 +8,4 @@ flask-bcrypt==1.0.1
 pyjwt==2.6.0
 gunicorn==20.1.0
 Werkzeug==2.3.7
+python-json-logger==2.0.7

auth-domain/user-manager/src/__init__.py

@@ -1,36 +1,64 @@
+import logging
+import logging.config
 import os
+import sys
 
-from flask import Flask
+from flask import Flask, request
 from flask_bcrypt import Bcrypt
 from flask_cors import CORS
 from flask_sqlalchemy import SQLAlchemy
 
-# instantiate the db
 db = SQLAlchemy()
 cors = CORS()
 bcrypt = Bcrypt()
 
+logging_config = {
+    "version": 1,
+    "formatters": {
+        "json": {
+            "class": "pythonjsonlogger.jsonlogger.JsonFormatter",
+            "format": "%(asctime)s %(process)s %(levelname)s",
+        }
+    },
+    "handlers": {
+        "console": {
+            "level": "DEBUG",
+            "class": "logging.StreamHandler",
+            "formatter": "json",
+            "stream": sys.stderr,
+        }
+    },
+    "root": {"level": "DEBUG", "handlers": ["console"], "propagate": True},
+}
+
+logging.config.dictConfig(logging_config)
+logger = logging.getLogger(__name__)
+
 
 def create_app(script_info=None):
-    # instantiate the app
     app = Flask(__name__)
 
-    # set config
     app_settings = os.getenv("APP_SETTINGS")
     app.config.from_object(app_settings)
 
-    # set up extensions
     db.init_app(app)
     cors.init_app(app, resources={r"*": {"origins": "*"}})
 
-    # register api
     from src.api import api
 
     api.init_app(app)
 
-    # shell context for flask cli
     @app.shell_context_processor
     def ctx():
         return {"app": app, "db": db}
 
+    @app.after_request
+    def log_info(response):
+        logging_dict = {}
+        logging_dict["request"] = {"path": request.path, "status": request.method}
+        logging_dict["response"] = {"status": response.status}
+        logging_dict["X-API-REQUEST-ID"] = request.headers.get("x-api-request-id")
+        logger.info(logging_dict)
+        return response
+
     return app

auth-domain/user-manager/src/api/models/users.py

@@ -29,7 +29,6 @@ class User(db.Model):
 
     @staticmethod
     def encode_token(user_id, token_type, airline=False):
-        print(f"encode_token(user_id={user_id}, token_type={token_type}):")
         if token_type == "access":
             seconds = current_app.config.get("ACCESS_TOKEN_EXPIRATION")
         else:

auth-domain/user-manager/src/tests/functional/test_auth.py

@@ -22,8 +22,6 @@ def test_user_registration(test_app, test_database):
         content_type="application/json",
     )
     data = json.loads(resp.data.decode())
-    print(data)
-    print(resp)
     assert resp.status_code == 201
     assert resp.content_type == "application/json"
     assert TEST_USERNAME in data["username"]

flights-domain/.env.dev.example

@@ -3,3 +3,4 @@ POSTGRES_PASS=password
 POSTGRES_DB=api_dev
 APP_SETTINGS=src.config.DevelopmentConfig
 ENVIRONMENT=dev
+API_DEBUG=true

flights-domain/.env.prod.example

@@ -3,3 +3,4 @@ POSTGRES_PASS=password
 POSTGRES_DB=api_prod
 APP_SETTINGS=src.config.ProductionConfig
 ENVIRONMENT=prod
+API_DEBUG=false

flights-domain/docker-compose.dev.yml

@@ -15,6 +15,9 @@ services:
     networks:
       - flights
       - subscriptions
+    logging:
+      options:
+        tag: dev-flights
 
   flights-db:
     extends:

flights-domain/docker-compose.yml

@@ -12,6 +12,9 @@ services:
     networks:
       - flights
       - subscriptions
+    logging:
+      options:
+        tag: flights
 
   flights-db:
     extends:

flights-domain/docker-template.yml

@@ -15,6 +15,10 @@ services:
       - PORT=5000
       - DATABASE_URL=postgresql://${POSTGRES_USER}:${POSTGRES_PASS}@flights-db/${POSTGRES_DB}
       - APP_SETTINGS=${APP_SETTINGS}
+    logging:
+      driver: gelf
+      options:
+        gelf-address: "udp://localhost:12201"
 
   flights-db:
     container_name: fids-flights_flights-db

flights-domain/flights-information/requirements.txt

@@ -4,4 +4,5 @@ psycopg2-binary==2.9.5
 pyjwt==2.6.0
 gunicorn==20.1.0
 sqlalchemy==2.0.22
-asyncreq==0.0.4
+asyncreq==0.0.5
+logmiddleware==0.0.3

flights-domain/flights-information/src/api/config.py

@@ -1,6 +1,7 @@
 import os
 
 TEST_TARGET = os.getenv("TEST_TARGET")
+API_DEBUG = os.getenv("API_DEBUG")
 
 if TEST_TARGET == "INTEGRATION":
     API_MESSAGES = "http://fids-subs-dev_subscriptions-api:5000/messages"

flights-domain/flights-information/src/api/cruds/flight.py

@@ -101,7 +101,6 @@ def update_flight(db: Session, update_data, id):
     db_flight = db.query(Flight).filter(Flight.id == id).first()
     if db_flight is None:
         raise KeyError
-    print(update_data)
     if db_flight.user_id != update_data["user_id"]:
         raise PermissionError
 

flights-domain/flights-information/src/api/main.py

@@ -1,9 +1,16 @@
+import logging
+
 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
+from logmiddleware import RouterLoggingMiddleware, logging_config
 
+from src.api.config import API_DEBUG
 from src.api.db import Base, engine
 from src.api.routes import flights, health
 
+logging.config.dictConfig(logging_config)
+logger = logging.getLogger(__name__)
+
 Base.metadata.create_all(bind=engine)
 
 
@@ -23,3 +30,4 @@ app.add_middleware(
     allow_methods=["POST", "GET", "PUT", "DELETE", "OPTIONS"],
     allow_headers=["*"],
 )
+app.add_middleware(RouterLoggingMiddleware, logger=logger, api_debug=API_DEBUG)

flights-domain/flights-information/src/api/routes/flights.py

@@ -1,7 +1,7 @@
-from typing import Optional
+from typing import Annotated, Optional
 
 from asyncreq import request
-from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException
+from fastapi import APIRouter, BackgroundTasks, Depends, Header, HTTPException, Request
 from sqlalchemy.orm import Session
 
 from src.api.config import API_MESSAGES
@@ -33,7 +33,9 @@ async def update_flight(
     id: int,
     update: FlightUpdate,
     background_tasks: BackgroundTasks,
+    req: Request,
     db: Session = Depends(get_db),
+    x_api_request_id: Annotated[str | None, Header()] = None,
 ):
     try:
         update_data = {
@@ -41,7 +43,6 @@ async def update_flight(
             for key, value in update.model_dump().items()
             if value is not None
         }
-        print(update_data)
         db_flight = flight_crud.update_flight(db=db, id=id, update_data=update_data)
     except PermissionError:
         raise HTTPException(status_code=401, detail="Unauthorized")
@@ -59,7 +60,8 @@ async def update_flight(
     msg["flight_code"] = db_flight.flight_code
     msg["origin"] = db_flight.origin
     msg["destination"] = db_flight.destination
-    background_tasks.add_task(request, API_MESSAGES, "POST", json=msg)
+    header = {"x-api-request-id": x_api_request_id}
+    background_tasks.add_task(request, API_MESSAGES, "POST", json=msg, headers=header)
     return db_flight
 
 

gateway/.env.dev.example

@@ -0,0 +1 @@
+API_DEBUG=true

gateway/.env.prod.example

@@ -0,0 +1 @@
+API_DEBUG=false

gateway/docker-compose.dev.yml

@@ -14,6 +14,9 @@ services:
       - flights
       - gateway
       - subscriptions
+    logging:
+      options:
+        tag: dev-gateway
 
 networks:
   auth:

gateway/docker-compose.yml

@@ -1,5 +1,5 @@
 version: '3.8'
-name: fids-gateway-dev
+name: fids-gateway
 
 services:
   api-gw:
@@ -14,10 +14,8 @@ services:
       - gateway
       - subscriptions
     logging:
-      driver: gelf
       options:
-        gelf-address: "udp://localhost:12201"
-        labels: gateway
+        tag: gateway
 
 networks:
   auth:

gateway/docker-template.yml

@@ -13,3 +13,8 @@ services:
     environment:
       - TEST_TARGET=${TEST_TARGET}
       - APP_SETTINGS=${APP_SETTINGS}
+      - API_DEBUG=${API_DEBUG}
+    logging:
+      driver: gelf
+      options:
+        gelf-address: "udp://localhost:12201"

gateway/logging.conf

@@ -1,21 +0,0 @@
-[loggers]
-keys=root
-
-[handlers]
-keys=consoleHandler
-
-[formatters]
-keys=normalFormatter
-
-[logger_root]
-level=INFO
-handlers=consoleHandler
-
-[handler_consoleHandler]
-class=StreamHandler
-level=DEBUG
-formatter=normalFormatter
-args=(sys.stdout,)
-
-[formatter_normalFormatter]
-format=%(asctime)s loglevel=%(levelname)-6s logger=%(name)s %(funcName)s() L%(lineno)-4d %(message)s

gateway/requirements.txt

@@ -3,5 +3,5 @@ fastapi[all]==0.103.2
 pyjwt==2.6.0
 gunicorn==20.1.0
 requests==2.31.0
-asyncreq==0.0.4
-graypy
+asyncreq==0.0.5
+logmiddleware==0.0.3

gateway/src/api/config.py

@@ -1,6 +1,7 @@
 import os
 
 TEST_TARGET = os.getenv("TEST_TARGET")
+API_DEBUG = os.getenv("API_DEBUG")
 
 if TEST_TARGET == "INTEGRATION":
     API_USERS = "http://fids-auth-dev_auth-api:5000/users"

gateway/src/api/log.py

@@ -0,0 +1,150 @@
+import json
+import logging
+import sys
+import time
+from typing import Callable
+from uuid import uuid4
+
+from fastapi import FastAPI, Request, Response
+from starlette.middleware.base import BaseHTTPMiddleware
+from starlette.types import Message
+
+logging_config = {
+    "version": 1,
+    "formatters": {
+        "json": {
+            "class": "pythonjsonlogger.jsonlogger.JsonFormatter",
+            "format": "%(asctime)s %(process)s %(levelname)s",
+        }
+    },
+    "handlers": {
+        "console": {
+            "level": "DEBUG",
+            "class": "logging.StreamHandler",
+            "formatter": "json",
+            "stream": sys.stderr,
+        }
+    },
+    "root": {"level": "DEBUG", "handlers": ["console"], "propagate": True},
+}
+
+
+class RouterLoggingMiddleware(BaseHTTPMiddleware):
+    def __init__(
+        self, app: FastAPI, *, logger: logging.Logger, api_debug: bool = False
+    ) -> None:
+        self._logger = logger
+        self.api_debug = api_debug
+        super().__init__(app)
+
+    async def dispatch(self, request: Request, call_next: Callable) -> Response:
+        request_header = request.headers.get("x-api-request-id")
+        if request_header is not None:
+            request_id = request_header
+        else:
+            request_id: str = str(uuid4())
+
+        logging_dict = {"X-API-REQUEST-ID": request_id}
+
+        if self.api_debug:
+            await self.set_body(request)
+
+        response, response_dict = await self._log_response(
+            call_next, request, request_id
+        )
+        request_dict = await self._log_request(request)
+        logging_dict["request"] = request_dict
+        logging_dict["response"] = response_dict
+
+        self._logger.info(logging_dict)
+
+        return response
+
+    async def set_body(self, request: Request):
+        _receive = await request._receive()
+
+        async def receive() -> Message:
+            return _receive
+
+        request._receive = receive
+
+    async def _log_request(self, request: Request) -> str:
+        path = request.url.path
+        if request.query_params:
+            path += f"?{request.query_params}"
+
+        request_logging = {
+            "method": request.method,
+            "path": path,
+            "ip": request.client.host,
+        }
+
+        if self.api_debug:
+            try:
+                body = await request.json()
+                request_logging["body"] = body
+            except ValueError:
+                body = None
+
+        return request_logging
+
+    async def _log_response(
+        self, call_next: Callable, request: Request, request_id: str
+    ) -> Response:
+        start_time = time.perf_counter()
+        response = await self._execute_request(call_next, request, request_id)
+        finish_time = time.perf_counter()
+
+        overall_status = "successful" if response.status_code < 400 else "failed"
+        execution_time = finish_time - start_time
+
+        response_logging = {
+            "status": overall_status,
+            "status_code": response.status_code,
+            "time_taken": f"{execution_time:0.4f}s",
+        }
+
+        if self.api_debug:
+            resp_body = [
+                section async for section in response.__dict__["body_iterator"]
+            ]
+            response.__setattr__("body_iterator", AsyncIteratorWrapper(resp_body))
+
+            try:
+                resp_body = json.loads(resp_body[0].decode())
+            except ValueError:
+                resp_body = str(resp_body)
+
+            response_logging["body"] = resp_body
+
+        return response, response_logging
+
+    async def _execute_request(
+        self, call_next: Callable, request: Request, request_id: str
+    ) -> Response:
+        try:
+            request.state.request_id = request_id
+            response: Response = await call_next(request)
+
+            response.headers["X-API-Request-ID"] = request_id
+            return response
+
+        except Exception as e:
+            self._logger.exception(
+                {"path": request.url.path, "method": request.method, "reason": e}
+            )
+
+
+class AsyncIteratorWrapper:
+    def __init__(self, obj):
+        self._it = iter(obj)
+
+    def __aiter__(self):
+        return self
+
+    async def __anext__(self):
+        try:
+            value = next(self._it)
+        except StopIteration:
+            raise StopAsyncIteration
+        return value

gateway/src/api/main.py

@@ -2,10 +2,16 @@ import logging
 
 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
+from logmiddleware import RouterLoggingMiddleware, logging_config
 
+from src.api.config import API_DEBUG
 from src.api.routes import auth, flights, health, notifications, subscriptions, users
 
-logging.config.fileConfig("logging.conf", disable_existing_loggers=False)
+# from src.api.log import RouterLoggingMiddleware, logging_config
+
+
+logging.config.dictConfig(logging_config)
+logger = logging.getLogger(__name__)
 
 
 app = FastAPI(title="Flights Information API")
@@ -28,3 +34,4 @@ app.add_middleware(
     allow_methods=["POST", "GET", "PUT", "DELETE", "OPTIONS"],
     allow_headers=["*"],
 )
+app.add_middleware(RouterLoggingMiddleware, logger=logger, api_debug=API_DEBUG)

gateway/src/api/routes/auth.py

@@ -1,7 +1,7 @@
 from typing import Annotated
 
 from asyncreq import request
-from fastapi import APIRouter, Header, HTTPException
+from fastapi import APIRouter, Header, HTTPException, Request
 
 from src.api.config import API_AUTH
 from src.api.schemas.auth import RefreshToken, Token
@@ -11,9 +11,11 @@ router = APIRouter()
 
 
 @router.post("/register", response_model=UserMin)
-async def register(user: UserRegister):
+async def register(user: UserRegister, req: Request):
+    request_id = req.state.request_id
+    header = {"x-api-request-id": request_id}
     (response, status, _) = await request(
-        f"{API_AUTH}/register", "POST", json=user.model_dump()
+        f"{API_AUTH}/register", "POST", json=user.model_dump(), headers=header
     )
     if status < 200 or status > 204:
         raise HTTPException(status_code=status, detail=response)
@@ -21,9 +23,11 @@ async def register(user: UserRegister):
 
 
 @router.post("/login", response_model=Token)
-async def login(user: UserLogin):
+async def login(user: UserLogin, req: Request):
+    request_id = req.state.request_id
+    header = {"x-api-request-id": request_id}
     (response, status, _) = await request(
-        f"{API_AUTH}/login", "POST", json=user.model_dump()
+        f"{API_AUTH}/login", "POST", json=user.model_dump(), headers=header
     )
     if status < 200 or status > 204:
         raise HTTPException(status_code=status, detail=response)
@@ -31,9 +35,11 @@ async def login(user: UserLogin):
 
 
 @router.post("/refresh", response_model=Token)
-async def refresh(token: RefreshToken):
+async def refresh(token: RefreshToken, req: Request):
+    request_id = req.state.request_id
+    header = {"x-api-request-id": request_id}
     (response, status, _) = await request(
-        f"{API_AUTH}/refresh", "POST", json=token.model_dump()
+        f"{API_AUTH}/refresh", "POST", json=token.model_dump(), headers=header
     )
     if status < 200 or status > 204:
         raise HTTPException(status_code=status, detail=response)
@@ -41,8 +47,12 @@ async def refresh(token: RefreshToken):
 
 
 @router.get("/status", response_model=UserMin)
-async def status(authorization: Annotated[str | None, Header()] = None):
-    header = {"Authorization": authorization if authorization is not None else ""}
+async def status(req: Request, authorization: Annotated[str | None, Header()] = None):
+    request_id = req.state.request_id
+    header = {
+        "Authorization": authorization if authorization is not None else "",
+        "x-api-request-id": request_id,
+    }
     (response, status, _) = await request(f"{API_AUTH}/status", "GET", headers=header)
     if status < 200 or status > 204:
         raise HTTPException(status_code=status, detail=response)

gateway/src/api/routes/flights.py

@@ -1,7 +1,7 @@
 from typing import Annotated, Optional
 
 from asyncreq import request
-from fastapi import APIRouter, Header, HTTPException
+from fastapi import APIRouter, Header, HTTPException, Request
 
 from src.api.config import API_FLIGHTS
 from src.api.routes.auth import status as checkAuth
@@ -11,8 +11,13 @@ router = APIRouter()
 
 
 @router.get("/{id}", response_model=Flight)
-async def get_flight_by_id(id: int):
-    (response, status, _) = await request(f"{API_FLIGHTS}/{id}", "GET")
+async def get_flight_by_id(
+    id: int,
+    req: Request,
+):
+    request_id = req.state.request_id
+    header = {"x-api-request-id": request_id}
+    (response, status, _) = await request(f"{API_FLIGHTS}/{id}", "GET", headers=header)
     if status < 200 or status > 204:
         raise HTTPException(status_code=status, detail=response)
     return response
@@ -20,12 +25,18 @@ async def get_flight_by_id(id: int):
 
 @router.post("", response_model=Flight)
 async def create_flight(
-    flight: FlightCreate, authorization: Annotated[str | None, Header()] = None
+    flight: FlightCreate,
+    req: Request,
+    authorization: Annotated[str | None, Header()] = None,
 ):
-    auth = await checkAuth(authorization)
+    auth = await checkAuth(req, authorization)
     flight_data = flight.model_dump()
     flight_data["user_id"] = auth["id"]
-    (response, status, _) = await request(f"{API_FLIGHTS}", "POST", json=flight_data)
+    request_id = req.state.request_id
+    header = {"x-api-request-id": request_id}
+    (response, status, _) = await request(
+        f"{API_FLIGHTS}", "POST", json=flight_data, headers=header
+    )
     if status < 200 or status > 204:
         raise HTTPException(status_code=status, detail=response)
     return response
@@ -35,12 +46,17 @@ async def create_flight(
 async def update_flight(
     id: int,
     flight_update: FlightUpdate,
+    req: Request,
     authorization: Annotated[str | None, Header()] = None,
 ):
-    auth = await checkAuth(authorization)
+    auth = await checkAuth(req, authorization)
     update = flight_update.model_dump()
     update["user_id"] = auth["id"]
-    (response, status, _) = await request(f"{API_FLIGHTS}/{id}", "PATCH", json=update)
+    request_id = req.state.request_id
+    header = {"x-api-request-id": request_id}
+    (response, status, _) = await request(
+        f"{API_FLIGHTS}/{id}", "PATCH", json=update, headers=header
+    )
     if status < 200 or status > 204:
         raise HTTPException(status_code=status, detail=response)
     return response
@@ -48,6 +64,7 @@ async def update_flight(
 
 @router.get("", response_model=list[Flight])
 async def get_flights(
+    req: Request,
     origin: Optional[str] = None,
     destination: Optional[str] = None,
     lastUpdated: Optional[str] = None,
@@ -62,7 +79,11 @@ async def get_flights(
         query["lastUpdated"] = lastUpdated
     if future:
         query["future"] = future
-    (response, status, _) = await request(f"{API_FLIGHTS}", "GET", query=query)
+    request_id = req.state.request_id
+    header = {"x-api-request-id": request_id}
+    (response, status, _) = await request(
+        f"{API_FLIGHTS}", "GET", query=query, headers=header
+    )
     if status < 200 or status > 204:
         raise HTTPException(status_code=status, detail=response)
     return response

gateway/src/api/routes/notifications.py

@@ -1,5 +1,5 @@
 from asyncreq import request
-from fastapi import APIRouter, HTTPException
+from fastapi import APIRouter, HTTPException, Request
 
 from src.api.config import API_NOTIFICATIONS
 from src.api.schemas.notification import Update as Message
@@ -8,10 +8,12 @@ router = APIRouter()
 
 
 @router.post("")
-async def receive_message(message: Message):
+async def receive_message(message: Message, req: Request):
     print(message.model_dump())
+    request_id = req.state.request_id
+    header = {"x-api-request-id": request_id}
     (response, status, _) = await request(
-        f"{API_NOTIFICATIONS}", "POST", json=message.model_dump()
+        f"{API_NOTIFICATIONS}", "POST", json=message.model_dump(), headers=header
     )
     if status < 200 or status > 204:
         raise HTTPException(status_code=status, detail=response)

gateway/src/api/routes/subscriptions.py

@@ -1,7 +1,7 @@
 from typing import Annotated
 
 from asyncreq import request
-from fastapi import APIRouter, Header, HTTPException
+from fastapi import APIRouter, Header, HTTPException, Request
 
 from src.api.config import API_SUBSCRIPTIONS
 from src.api.routes.auth import status as checkAuth
@@ -12,11 +12,15 @@ router = APIRouter()
 
 @router.post("")
 async def create_subscription(
-    subscription: Subscription, authorization: Annotated[str | None, Header()] = None
+    subscription: Subscription,
+    req: Request,
+    authorization: Annotated[str | None, Header()] = None,
 ):
-    await checkAuth(authorization)
+    await checkAuth(req, authorization)
+    request_id = req.state.request_id
+    header = {"x-api-request-id": request_id}
     (response, status, _) = await request(
-        f"{API_SUBSCRIPTIONS}", "POST", json=subscription.model_dump()
+        f"{API_SUBSCRIPTIONS}", "POST", json=subscription.model_dump(), headers=header
     )
     if status < 200 or status > 204:
         raise HTTPException(status_code=status, detail=response)

gateway/src/api/routes/users.py

@@ -1,5 +1,5 @@
 from asyncreq import request
-from fastapi import APIRouter, HTTPException
+from fastapi import APIRouter, HTTPException, Request
 
 from src.api.config import API_USERS
 from src.api.schemas.user import User, UserRegister
@@ -8,17 +8,21 @@ router = APIRouter()
 
 
 @router.get("", response_model=list[User])
-async def get_users():
-    (response, status, _) = await request(f"{API_USERS}", "GET")
+async def get_users(req: Request):
+    request_id = req.state.request_id
+    header = {"x-api-request-id": request_id}
+    (response, status, _) = await request(f"{API_USERS}", "GET", headers=header)
     if status < 200 or status > 204:
         raise HTTPException(status_code=status, detail=response)
     return response
 
 
 @router.post("", response_model=User)
-async def create_users(user: UserRegister):
+async def create_users(user: UserRegister, req: Request):
+    request_id = req.state.request_id
+    header = {"x-api-request-id": request_id}
     (response, status, _) = await request(
-        f"{API_USERS}", "POST", json=user.dump_model()
+        f"{API_USERS}", "POST", json=user.dump_model(), headers=header
     )
     if status < 200 or status > 204:
         raise HTTPException(status_code=status, detail=response)
@@ -26,17 +30,21 @@ async def create_users(user: UserRegister):
 
 
 @router.get("/{id}", response_model=User)
-async def get_user(id: str):
-    (response, status, _) = await request(f"{API_USERS}/{id}", "GET")
+async def get_user(id: str, req: Request):
+    request_id = req.state.request_id
+    header = {"x-api-request-id": request_id}
+    (response, status, _) = await request(f"{API_USERS}/{id}", "GET", headers=header)
     if status < 200 or status > 204:
         raise HTTPException(status_code=status, detail=response)
     return response
 
 
 @router.put("/{id}", response_model=User)
-async def update_user(user: UserRegister):
+async def update_user(user: UserRegister, req: Request):
+    request_id = req.state.request_id
+    header = {"x-api-request-id": request_id}
     (response, status, _) = await request(
-        f"{API_USERS}/{id}", "PUT", json=user.model_dump()
+        f"{API_USERS}/{id}", "PUT", json=user.model_dump(), headers=header
     )
     if status < 200 or status > 204:
         raise HTTPException(status_code=status, detail=response)
@@ -44,8 +52,10 @@ async def update_user(user: UserRegister):
 
 
 @router.delete("/{id}", response_model=User)
-async def delete_user():
-    (response, status, _) = await request(f"{API_USERS}/{id}", "DELETE")
+async def delete_user(req: Request):
+    request_id = req.state.request_id
+    header = {"x-api-request-id": request_id}
+    (response, status, _) = await request(f"{API_USERS}/{id}", "DELETE", headers=header)
     if status < 200 or status > 204:
         raise HTTPException(status_code=status, detail=response)
     return response

observability/elk/logstash/pipeline/logstash.conf

@@ -6,11 +6,29 @@ input {
 	}
 }
 
+filter {
+	mutate {
+      remove_field => [ "host" ]
+    }
+    json {
+        source => "message"
+        target => "jsoncontent"
+    }
+    if [jsoncontent][response][body] {
+        mutate {
+            add_field => { "data" => "%{[jsoncontent][response][body]}" }
+            remove_field => [ "[jsoncontent][response][body]" ]
+        }
+    }
+}
+
+
 output {
 	elasticsearch {
-		index => "logs-%{+YYYY.MM.dd}"
+		index => "logs-%{tag}-%{+YYYY.MM.dd}"
 		hosts => "elasticsearch:9200"
 		user => "logstash_internal"
 		password => "${LOGSTASH_INTERNAL_PASSWORD}"
+        action => "create"
 	}
 }

run.sh

@@ -23,24 +23,49 @@ done
 if [ -n "$domain" ] && [ -n "$down" ]; then
     case "$domain" in
         'auth')
+            if [ -n "$tests" ]; then
+                export API_IMAGE=$USER/user-manager:test
+                docker compose -f auth-domain/docker-compose.dev.yml --env-file auth-domain/.env.dev down
+            else
                 export API_IMAGE=$USER/user-manager:prod
                 docker compose -f auth-domain/docker-compose.yml --env-file auth-domain/.env.prod down
+            fi
             ;;
         'flights')
+            if [ -n "$tests" ]; then
+                export API_IMAGE=$USER/flights-information:test
+                docker compose -f flights-domain/docker-compose.dev.yml --env-file flights-domain/.env.dev down
+            else
                 export API_IMAGE=$USER/flights-information:prod
                 docker compose -f flights-domain/docker-compose.yml --env-file flights-domain/.env.prod down
+            fi
             ;;
         'gateway')
+            if [ -n "$tests" ]; then
+                export API_IMAGE=$USER/gateway:test
+                docker compose -f gateway/docker-compose.dev.yml --env-file gateway/.env.dev down
+            else
                 export API_IMAGE=$USER/gateway:prod
-            docker compose -f gateway/docker-compose.yml down
+                docker compose -f gateway/docker-compose.yml --env-file gateway/.env.prod down
+            fi
             ;;
         'screen')
+            if [ -n "$tests" ]; then
+                export CLIENT_IMAGE=$USER/screen-client:test
+                docker compose -f screen-domain/docker-compose.dev.yml down
+            else
                 export CLIENT_IMAGE=$USER/screen-client:prod
                 docker compose -f screen-domain/docker-compose.yml down
+            fi
             ;;
         'browser')
+            if [ -n "$tests" ]; then
+                export CLIENT_IMAGE=$USER/browser-client:test
+                docker compose -f browser-domain/docker-compose.dev.yml down
+            else
                 export CLIENT_IMAGE=$USER/browser-client:prod
                 docker compose -f browser-domain/docker-compose.yml down
+            fi
             ;;
         'elk')
             export ELASTICSEARCH_IMAGE=$USER/elasticsearch:prod
@@ -52,8 +77,13 @@ if [ -n "$domain" ] && [ -n "$down" ]; then
             docker compose -f observability/docker-compose.yml -f observability/elk/extensions/curator/curator-compose.yml -f observability/elk/extensions/heartbeat/heartbeat-compose.yml --env-file observability/.env.prod down
             ;;
         'subscription')
+            if [ -n "$tests" ]; then
+                export API_IMAGE=$USER/subs-manager:test
+                docker compose -f subscription-domain/docker-compose.dev.yml --env-file subscription-domain/.env.dev down
+            else
                 export API_IMAGE=$USER/subs-manager:prod
                 docker compose -f subscription-domain/docker-compose.yml --env-file subscription-domain/.env.prod down
+            fi
             ;;
         *) exit 1 ;;
     esac
@@ -118,22 +148,24 @@ elif [ -n "$domain" ] && [ -z "$down" ]; then
             if [ -n "$tests" ]; then
                 docker build gateway -f gateway/Dockerfile.test -t $USER/gateway:test
                 export API_IMAGE=$USER/gateway:test
-                docker compose -f gateway/docker-compose.dev.yml down
-                docker compose -f gateway/docker-compose.dev.yml up -d
+                docker compose -f gateway/docker-compose.dev.yml --env-file gateway/.env.dev down
+                docker compose -f gateway/docker-compose.dev.yml --env-file gateway/.env.dev up -d
             else
                 export API_IMAGE=$USER/gateway:prod
-                docker compose -f gateway/docker-compose.yml down
-                docker compose -f gateway/docker-compose.yml up -d
+                docker compose -f gateway/docker-compose.yml --env-file gateway/.env.prod down
+                docker compose -f gateway/docker-compose.yml --env-file gateway/.env.prod up -d
             fi
             ;;
         'screen')
             if [ -n "$tests" ]; then
                 docker build screen-domain -f screen-domain/Dockerfile.test -t $USER/screen-client:test
                 export CLIENT_IMAGE=$USER/screen-client:test
+                docker compose -f screen-domain/docker-compose.dev.yml down
                 docker compose -f screen-domain/docker-compose.dev.yml up -d
             else
                 docker build screen-domain -f screen-domain/Dockerfile.prod --build-arg "REACT_APP_ORIGIN=$REACT_APP_ORIGIN" -t $USER/screen-client:prod
                 export CLIENT_IMAGE=$USER/screen-client:prod
+                docker compose -f screen-domain/docker-compose.yml down
                 docker compose -f screen-domain/docker-compose.yml up -d
             fi
             ;;
@@ -141,10 +173,12 @@ elif [ -n "$domain" ] && [ -z "$down" ]; then
             if [ -n "$tests" ]; then
                 docker build browser-domain -f browser-domain/Dockerfile.test -t $USER/browser-client:test
                 export CLIENT_IMAGE=$USER/browser-client:test
+                docker compose -f browser-domain/docker-compose.dev.yml down
                 docker compose -f browser-domain/docker-compose.dev.yml up -d
             else
                 docker build browser-domain -f browser-domain/Dockerfile.prod -t $USER/browser-client:prod
                 export CLIENT_IMAGE=$USER/browser-client:prod
+                docker compose -f browser-domain/docker-compose.yml down
                 docker compose -f browser-domain/docker-compose.yml up -d
             fi
             ;;
@@ -176,7 +210,7 @@ elif [ -n "$down" ]; then
     export API_IMAGE=slococo/subs-manager:prod
     docker compose -f subscription-domain/docker-compose.yml --env-file subscription-domain/.env.prod down
     export API_IMAGE=$USER/gateway:prod
-    docker compose -f gateway/docker-compose.yml down
+    docker compose -f gateway/docker-compose.yml --env-file gateway/.env.prod down
 
     export CLIENT_IMAGE=$USER/screen-client:prod
     docker compose -f screen-domain/docker-compose.yml down
@@ -207,8 +241,8 @@ else
     docker compose -f subscription-domain/docker-compose.yml --env-file subscription-domain/.env.prod down
     docker compose -f subscription-domain/docker-compose.yml --env-file subscription-domain/.env.prod up -d
     export API_IMAGE=$USER/gateway:prod
-    docker compose -f gateway/docker-compose.yml down
-    docker compose -f gateway/docker-compose.yml up -d
+    docker compose -f gateway/docker-compose.yml --env-file gateway/.env.prod down
+    docker compose -f gateway/docker-compose.yml --env-file gateway/.env.prod up -d
 
     export CLIENT_IMAGE=$USER/screen-client:prod
     docker compose -f screen-domain/docker-compose.yml up -d

subscription-domain/.env.dev.example

@@ -3,3 +3,4 @@ POSTGRES_PASS=password
 POSTGRES_DB=api_dev
 APP_SETTINGS=src.config.DevelopmentConfig
 TOKEN=3275588851:AT36AGy_BChQUuCq2M6d2UrY5CSWtZe45gV
+API_DEBUG=true

subscription-domain/.env.prod.example

@@ -3,3 +3,4 @@ POSTGRES_PASS=password
 POSTGRES_DB=api_prod
 APP_SETTINGS=src.config.ProductionConfig
 TOKEN=3275588851:AT36AGy_BChQUuCq2M6d2UrY5CSWtZe45gV
+API_DEBUG=false

subscription-domain/docker-compose.dev.yml

@@ -14,6 +14,9 @@ services:
         condition: service_healthy
     networks:
       - subscriptions
+    logging:
+      options:
+        tag: dev-subs
 
   subscriptions-db:
     extends:

subscription-domain/docker-compose.yml

@@ -11,6 +11,9 @@ services:
         condition: service_healthy
     networks:
       - subscriptions
+    logging:
+      options:
+        tag: subs
 
   subscriptions-db:
     extends:

subscription-domain/docker-template.yml

@@ -16,6 +16,10 @@ services:
       - DATABASE_URL=postgresql://${POSTGRES_USER}:${POSTGRES_PASS}@subscriptions-db/${POSTGRES_DB}
       - APP_SETTINGS=${APP_SETTINGS}
       - TOKEN=${TOKEN}
+    logging:
+      driver: gelf
+      options:
+        gelf-address: "udp://localhost:12201"
 
   subscriptions-db:
     container_name: fids-subs_subscriptions-db

subscription-domain/subscription-manager/requirements.txt

@@ -4,4 +4,5 @@ psycopg2-binary==2.9.5
 pyjwt==2.6.0
 gunicorn==20.1.0
 sqlalchemy==2.0.22
-asyncreq==0.0.4
+asyncreq==0.0.5
+logmiddleware==0.0.3

subscription-domain/subscription-manager/src/api/config.py

@@ -1 +1,4 @@
+import os
+
 API_FLIGHTS = "http://fids_flights_api:5000/flights"
+API_DEBUG = os.getenv("API_DEBUG")

subscription-domain/subscription-manager/src/api/main.py

@@ -1,9 +1,16 @@
+import logging
+
 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
+from logmiddleware import RouterLoggingMiddleware, logging_config
 
+from src.api.config import API_DEBUG
 from src.api.db import Base, engine
 from src.api.routes import health, messages, notifications, subscriptions
 
+logging.config.dictConfig(logging_config)
+logger = logging.getLogger(__name__)
+
 Base.metadata.create_all(bind=engine)
 
 
@@ -25,3 +32,4 @@ app.add_middleware(
     allow_methods=["POST", "GET", "PUT", "DELETE", "OPTIONS"],
     allow_headers=["*"],
 )
+app.add_middleware(RouterLoggingMiddleware, logger=logger, api_debug=API_DEBUG)