Logging

Logging Stack

Grafana: Grafana is an open source visualization platform that allows developers to analyze logs. In order to see your progams logs in Grafana you are able to create dashboards which allow filtering by date and time, or by specific tags you applied to the logs. It supports many different data sources including Loki, which is what VeloSim uses for both development and production.

Loki: Loki is a log aggregator, meaning it efficiently stores and makes the collected logs queryable by the Grafana dashboard.

Promtail: Promtail is what is used in our application to scrape the logs written to logs.txt, then sending it to loki to be stored.

Production Deployment: The same logging stack (Grafana, Loki, Promtail) is deployed to both development and production environments. In production, these services are deployed via Ansible to an on-premise server, allowing all team members to access logs through Grafana's web interface without needing to run containers locally.

Log Retention: Logs are retained for 60 days (configured in grafana_logging/loki-config.yaml) to balance storage requirements with debugging needs.

Log Persistence: Logs are written to grafana_logging/logs.txt and persisted via Docker volume mounts. This ensures logs survive container restarts and are available for Promtail to scrape and send to Loki.

A significant amount of detail is logged automatically:

• API requests and response codes (but not including the requesting user)
• Startup/shutdown of the application
• Error traces which are logged to the console

Grafana queries can be used to interpret the logs.

Logging Library

At the moment we use Loki and promtail to produce more detailed logs from our app. Change are possible in the future if we see a better solution for these

This includes logs at the ERROR, INFO and WARNING levels from our application code, documenting significant events which occur in the apps.

This library can also collect metrics: for example, counts of significant events, or average response times.

Frontend Logs

We provide an endpoint /logs/frontend which the front-end can use to persist significant events that occur in the front-end to the back-end logs, allowing for a holistic analysis of the user's journey through the application.

Examples:

1. Basic Logging

from back.grafana_logging.logger import get_logger

logger = get_logger(__name__)
logger.debug("This is a debug message")
logger.info("Application started")
logger.warning("This is a warning")
logger.error("An error occurred")
logger.critical("Critical failure")

2. Structured Logging

logger = get_logger("user_service")
user_id = 12345
action = "login"

logger.info(
    f"User action completed | user_id={user_id} | action={action} | status=success"
)

3. Logging Errors

def example_error_logging_with_context() -> None:
    """Example: Logging errors with full context.

    Returns:
        None
    """
    logger = get_logger("database")

    try:
        # Simulating an error
        _ = 1 / 0
    except Exception as e:
        # Log with full traceback
        logger.error(
            f"Database operation failed: {e}",
            exc_info=True,  # Includes full stack trace
            extra={"operation": "insert", "table": "stations", "user_id": 12345},
        )

4. Frontend Logging

from fastapi import APIRouter, Body
from back.grafana_logging.logger import get_logger

router = APIRouter()
frontend_logger = get_logger("frontend")

@router.post("/api/v1/logs")
async def log_from_frontend(
    level: str = Body(...),
    message: str = Body(...),
    context: dict = Body(None)
):
    log_func = getattr(frontend_logger, level.lower(), frontend_logger.info)
    if context:
        context_str = ", ".join([f"{k}={v}" for k, v in context.items()])
        log_func(f"[Frontend] {message} | {context_str}")
    else:
        log_func(f"[Frontend] {message}")
    return {"status": "logged"}

5. Backend Logging

def example_backend_api_logging() -> List[str]:
    """Example: Logging in a FastAPI endpoint.

    Returns:
        List[str]: List of station names.
    """
    logger = get_logger("api.stations")

    # Log when endpoint is called
    logger.info("Fetching all stations from database")

    try:
        # Simulating database call
        stations = ["Station A", "Station B"]
        logger.info(f"Successfully retrieved {len(stations)} stations")
        return stations
    except Exception as e:
        logger.error(f"Failed to fetch stations: {e}", exc_info=True)
        raise

6. Simulator Logging

def example_simulator_logging() -> None:
    """Example: Logging in the simulator.

    Returns:
        None
    """
    logger = get_logger("simulator")

    # Log simulation start
    logger.info("Starting new simulation run")

    # Log simulation steps with data
    log_simulation_event(
        event_type="STEP",
        message="Vehicle moved to new position",
        data={
            "vehicle_id": 123,
            "position": (10.5, 20.3),
            "speed": 15.2,
            "timestamp": "2025-10-12T14:30:00",
        },
    )

    # Log simulation completion
    log_simulation_event(
        event_type="COMPLETE",
        message="Simulation finished successfully",
        data={"total_steps": 1000, "duration_seconds": 45.2},
    )