feat: add a Redis-backed driver API load balancer

.github/workflows/ci.yml

@@ -106,3 +106,72 @@ jobs:
       - name: Stop broker services
         if: always()
         run: docker compose -f tests/integration/docker-compose.yml down --volumes
+
+
+  load-balancer-integration:
+    name: Driver API load-balancer integration
+    runs-on: ubuntu-latest
+    timeout-minutes: 15
+
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v4
+
+      - name: Validate NGINX configuration
+        run: |
+          docker run --rm \
+            -v "$PWD/nginx/driver-location-load-balancer.conf:/etc/nginx/nginx.conf:ro" \
+            nginx:1.27.5-alpine nginx -t
+
+      - name: Start three driver API replicas behind the gateway
+        env:
+          EXPOSE_INSTANCE_ID: "true"
+        run: |
+          docker compose up --build --detach --scale driver-location-api=3 api-gateway
+
+      - name: Verify readiness through the public gateway
+        run: |
+          for attempt in $(seq 1 45); do
+            if curl --fail --silent --show-error http://localhost:8000/driver-location/ready > /dev/null; then
+              exit 0
+            fi
+            sleep 2
+          done
+          echo "Gateway never became ready" >&2
+          docker compose ps
+          docker compose logs --no-color driver-location-api api-gateway redis
+          exit 1
+
+      - name: Verify the gateway reads shared Redis state
+        run: |
+          docker compose exec -T redis redis-cli HSET driver-locations ci-driver \
+            '{"driver_id":"ci-driver","lat":40.7128,"lon":-74.0060,"timestamp":"2026-01-01T00:00:00Z","status":"available"}'
+          curl --fail --silent --show-error http://localhost:8000/driver-location/drivers/ci-driver \
+            | python -c "import json, sys; assert json.load(sys.stdin)['driver_id'] == 'ci-driver'"
+
+      - name: Verify requests reach multiple replicas
+        run: |
+          for request in $(seq 1 18); do
+            curl --fail --silent --show-error --dump-header - --output /dev/null \
+              http://localhost:8000/driver-location/health \
+              | tr -d '\r' | awk -F': ' 'tolower($1) == "x-instance-id" {print $2}'
+          done | sort -u > /tmp/driver-api-replicas.txt
+          cat /tmp/driver-api-replicas.txt
+          test "$(wc -l < /tmp/driver-api-replicas.txt)" -ge 2
+
+      - name: Verify one backend loss does not take down readiness
+        run: |
+          docker ps --filter label=com.docker.compose.service=driver-location-api --format '{{.ID}}' \
+            | head -n 1 | xargs --no-run-if-empty docker stop
+          for attempt in $(seq 1 15); do
+            if curl --fail --silent --show-error http://localhost:8000/driver-location/ready > /dev/null; then
+              exit 0
+            fi
+            sleep 1
+          done
+          echo "Gateway did not remain ready after one replica stopped" >&2
+          exit 1
+
+      - name: Stop Compose services
+        if: always()
+        run: docker compose down --volumes

Dailylog.md

@@ -0,0 +1,9 @@
+# Daily Log
+
+## 2026-08-12 — Load-balancer gap recorded
+
+An infrastructure audit found that the Compose demo exposes one `driver-location-api` instance directly on host port 8000. The event-driven architecture has no local Layer-7 gateway, multi-replica API routing, or unhealthy-replica resilience demonstration.
+
+Tracking issue: [#17 — add a gateway load balancer for horizontally scaled APIs](https://github.com/CoreyLeath-code/Scalable-Event-Driven-Ride-Sharing-Platform/issues/17).
+
+This record does **not** claim a load balancer, horizontal scaling, resilience test, or benchmark result has been implemented. The issue defines the required implementation and validation scope.

README.md

@@ -176,6 +176,10 @@ The Compose profile starts the repository's root driver-location API and validat
 health endpoint. Kafka and the additional service boundaries remain architectural extension
 points; they are not started by this local demo profile.
 
+## Load-balanced driver-location API
+
+The public Compose endpoint at port `8000` is NGINX; it forwards to internal `driver-location-api` replicas using least-connections routing. `/driver-location/health` is a liveness probe, while `/driver-location/ready` verifies the configured Redis-backed driver store. The CI integration job validates the NGINX configuration, replica routing, shared-state read, and continued readiness after one replica stops; `EXPOSE_INSTANCE_ID=true` is limited to that test and is disabled by default.
+
 ## Event Flow
 
 ```text

api_router.py

@@ -49,7 +49,22 @@ async def get_driver_count():
 
 @router.get("/health")
 async def health_check():
-    """
-    Returns service health.
-    """
+    """Liveness probe: the API process can serve requests."""
+    return {"status": "OK", "service": "driver-location-service"}
+
+
+@router.get("/ready")
+async def readiness_check():
+    """Readiness probe: the configured driver-location store is reachable."""
+    if DRIVER_STORE is None:
+        raise HTTPException(503, "Driver store not initialized.")
+
+    try:
+        ready = DRIVER_STORE.is_ready()
+    except Exception as exc:
+        logger.error("Driver store readiness check failed: %s", exc)
+        raise HTTPException(503, "Driver store is unavailable.") from exc
+
+    if not ready:
+        raise HTTPException(503, "Driver store is unavailable.")
     return {"status": "OK", "service": "driver-location-service"}

docker-compose.yml

@@ -1,20 +1,52 @@
 services:
+  redis:
+    image: redis:7.4-alpine
+    restart: unless-stopped
+    command: ["redis-server", "--appendonly", "yes"]
+    volumes:
+      - driver-location-redis:/data
+    healthcheck:
+      test: ["CMD", "redis-cli", "ping"]
+      interval: 10s
+      timeout: 3s
+      retries: 5
+
   driver-location-api:
     build:
       context: .
-    ports:
-      - "8000:8000"
+    restart: unless-stopped
+    expose:
+      - "8000"
     environment:
       PYTHONUNBUFFERED: "1"
+      DRIVER_LOCATION_REDIS_URL: redis://redis:6379/0
+      EXPOSE_INSTANCE_ID: ${EXPOSE_INSTANCE_ID:-false}
+    depends_on:
+      redis:
+        condition: service_healthy
     healthcheck:
       test:
         [
           "CMD",
           "python",
           "-c",
-          "import urllib.request; urllib.request.urlopen('http://127.0.0.1:8000/driver-location/health', timeout=3)",
+          "import urllib.request; urllib.request.urlopen('http://127.0.0.1:8000/driver-location/ready', timeout=3)",
         ]
       interval: 15s
       timeout: 5s
       retries: 3
       start_period: 10s
+
+  api-gateway:
+    image: nginx:1.27.5-alpine
+    restart: unless-stopped
+    ports:
+      - "8000:8000"
+    volumes:
+      - ./nginx/driver-location-load-balancer.conf:/etc/nginx/nginx.conf:ro
+    depends_on:
+      driver-location-api:
+        condition: service_healthy
+
+volumes:
+  driver-location-redis:

location_store.py

@@ -1,75 +1,87 @@
-from models import DriverLocationEvent
-from utils import get_logger
-
-logger = get_logger("DriverLocationStore")
-
-
-class DriverLocationStore:
-    """
-    In-memory real-time driver store.
-
-    Responsibilities:
-    - Maintain driver availability
-    - Update driver coordinates
-    - Remove offline drivers
-    - Provide list of active drivers for Dispatch Service
-
-    This version is intentionally simple, but the interface allows
-    easy migration to Redis, MongoDB, or a geospatial index (H3).
-    """
-
-    def __init__(self):
-        # driver_id → DriverLocationEvent
-        self.drivers: dict[str, DriverLocationEvent] = {}
-
-    # ------------------------------------------------------------
-    # Core Operations
-    # ------------------------------------------------------------
-
-    def upsert_driver(self, event: DriverLocationEvent) -> None:
-        """
-        Add or update driver in the store.
-        """
-        self.drivers[event.driver_id] = event
-        logger.info(f"Updated driver {event.driver_id} at ({event.lat}, {event.lon})")
-
-    def remove_driver(self, driver_id: str) -> None:
-        """
-        Remove a driver from the store.
-        """
-        if driver_id in self.drivers:
-            del self.drivers[driver_id]
-            logger.info(f"Removed driver {driver_id}")
-
-    # ------------------------------------------------------------
-    # Retrieval
-    # ------------------------------------------------------------
-
-    def get_all_drivers(self) -> list[DriverLocationEvent]:
-        """
-        Return a list of all active drivers.
-        """
-        return list(self.drivers.values())
-
-    def get_driver(self, driver_id: str) -> DriverLocationEvent | None:
-        """
-        Fetch a single driver by ID.
-        """
-        return self.drivers.get(driver_id)
-
-    # ------------------------------------------------------------
-    # Debug Utilities
-    # ------------------------------------------------------------
-
-    def count(self) -> int:
-        """
-        Count active drivers.
-        """
-        return len(self.drivers)
-
-    def clear(self) -> None:
-        """
-        Remove all drivers (reset state).
-        """
-        self.drivers.clear()
-        logger.warning("Cleared all driver locations.")
+import redis
+
+from models import DriverLocationEvent
+from utils import get_logger
+
+logger = get_logger("DriverLocationStore")
+
+
+class DriverLocationStore:
+    """In-memory driver store for single-process local development."""
+
+    def __init__(self):
+        self.drivers: dict[str, DriverLocationEvent] = {}
+
+    def upsert_driver(self, event: DriverLocationEvent) -> None:
+        self.drivers[event.driver_id] = event
+        logger.info(f"Updated driver {event.driver_id} at ({event.lat}, {event.lon})")
+
+    def remove_driver(self, driver_id: str) -> None:
+        if driver_id in self.drivers:
+            del self.drivers[driver_id]
+            logger.info(f"Removed driver {driver_id}")
+
+    def get_all_drivers(self) -> list[DriverLocationEvent]:
+        return list(self.drivers.values())
+
+    def get_driver(self, driver_id: str) -> DriverLocationEvent | None:
+        return self.drivers.get(driver_id)
+
+    def count(self) -> int:
+        return len(self.drivers)
+
+    def clear(self) -> None:
+        self.drivers.clear()
+        logger.warning("Cleared all driver locations.")
+
+    def is_ready(self) -> bool:
+        return True
+
+
+class RedisDriverLocationStore:
+    """Redis-backed driver store for horizontally scaled API replicas."""
+
+    redis_key = "driver-locations"
+
+    def __init__(self, redis_url: str | None = None, client=None):
+        if client is not None:
+            self.client = client
+        elif redis_url is not None:
+            self.client = redis.Redis.from_url(redis_url, decode_responses=True)
+        else:
+            raise ValueError("redis_url is required when no Redis client is provided")
+
+    def upsert_driver(self, event: DriverLocationEvent) -> None:
+        self.client.hset(self.redis_key, event.driver_id, event.model_dump_json())
+        logger.info(f"Updated driver {event.driver_id} in Redis")
+
+    def remove_driver(self, driver_id: str) -> None:
+        self.client.hdel(self.redis_key, driver_id)
+
+    def get_all_drivers(self) -> list[DriverLocationEvent]:
+        return [
+            DriverLocationEvent.model_validate_json(serialized)
+            for serialized in self.client.hvals(self.redis_key)
+        ]
+
+    def get_driver(self, driver_id: str) -> DriverLocationEvent | None:
+        serialized = self.client.hget(self.redis_key, driver_id)
+        if serialized is None:
+            return None
+        return DriverLocationEvent.model_validate_json(serialized)
+
+    def count(self) -> int:
+        return self.client.hlen(self.redis_key)
+
+    def clear(self) -> None:
+        self.client.delete(self.redis_key)
+
+    def is_ready(self) -> bool:
+        return bool(self.client.ping())
+
+
+def create_driver_store(redis_url: str | None = None):
+    """Return a shared Redis store only when an explicit endpoint is configured."""
+    if redis_url:
+        return RedisDriverLocationStore(redis_url)
+    return DriverLocationStore()

main.py

@@ -1,10 +1,13 @@
+import os
+import socket
+
 from fastapi import FastAPI
 
 import api_router
 from api_router import router
 from consumer import DriverLocationConsumer
 from event_bus import EventBus
-from location_store import DriverLocationStore
+from location_store import create_driver_store
 from utils import get_logger
 
 # ------------------------------------------------------------
@@ -14,7 +17,7 @@
 logger = get_logger("DriverLocationMain")
 
 event_bus = EventBus()
-driver_store = DriverLocationStore()
+driver_store = create_driver_store(os.getenv("DRIVER_LOCATION_REDIS_URL"))
 
 app = FastAPI(
     title="Driver Location Service",
@@ -23,6 +26,15 @@
 )
 
 
+@app.middleware("http")
+async def optional_instance_id_header(request, call_next):
+    """Expose a container identity only when Compose smoke tests request it."""
+    response = await call_next(request)
+    if os.getenv("EXPOSE_INSTANCE_ID") == "true":
+        response.headers["X-Instance-ID"] = os.getenv("INSTANCE_ID", socket.gethostname())
+    return response
+
+
 # ------------------------------------------------------------
 # STARTUP SEQUENCE
 # ------------------------------------------------------------