diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index abd823f..e1cf5ca 100644 --- a/.github/workflows/ci.yml +++ b/.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 diff --git a/Dailylog.md b/Dailylog.md new file mode 100644 index 0000000..1c99816 --- /dev/null +++ b/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. diff --git a/README.md b/README.md index 355273f..3cd01b7 100644 --- a/README.md +++ b/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 diff --git a/api_router.py b/api_router.py index fa300dd..608c388 100644 --- a/api_router.py +++ b/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"} diff --git a/docker-compose.yml b/docker-compose.yml index c3dd137..596ce6d 100644 --- a/docker-compose.yml +++ b/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: diff --git a/location_store.py b/location_store.py index ca2f47f..74233eb 100644 --- a/location_store.py +++ b/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() diff --git a/main.py b/main.py index c110557..0de5e83 100644 --- a/main.py +++ b/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 # ------------------------------------------------------------ diff --git a/nginx/driver-location-load-balancer.conf b/nginx/driver-location-load-balancer.conf new file mode 100644 index 0000000..bfd041c --- /dev/null +++ b/nginx/driver-location-load-balancer.conf @@ -0,0 +1,68 @@ +worker_processes auto; + +events { + worker_connections 1024; +} + +http { + log_format upstream '$remote_addr "$request" $status ' + 'upstream=$upstream_addr upstream_status=$upstream_status ' + 'request_time=$request_time upstream_time=$upstream_response_time'; + access_log /var/log/nginx/access.log upstream; + + resolver 127.0.0.11 valid=10s ipv6=off; + + upstream driver_location_backends { + zone driver_location_backends 64k; + least_conn; + server driver-location-api:8000 resolve; + keepalive 32; + } + + server { + listen 8000; + server_name _; + + location = /nginx_status { + stub_status; + access_log off; + } + + location = /driver-location/health { + proxy_pass http://driver_location_backends; + proxy_http_version 1.1; + proxy_set_header Host $host; + proxy_set_header X-Real-IP $remote_addr; + proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; + proxy_set_header X-Forwarded-Proto $scheme; + proxy_next_upstream error timeout http_502 http_503 http_504; + proxy_next_upstream_tries 2; + proxy_connect_timeout 2s; + proxy_read_timeout 5s; + } + + location = /driver-location/ready { + proxy_pass http://driver_location_backends; + proxy_http_version 1.1; + proxy_set_header Host $host; + proxy_set_header X-Real-IP $remote_addr; + proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; + proxy_set_header X-Forwarded-Proto $scheme; + proxy_next_upstream error timeout http_502 http_503 http_504; + proxy_next_upstream_tries 2; + proxy_connect_timeout 2s; + proxy_read_timeout 5s; + } + + location / { + proxy_pass http://driver_location_backends; + proxy_http_version 1.1; + proxy_set_header Host $host; + proxy_set_header X-Real-IP $remote_addr; + proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; + proxy_set_header X-Forwarded-Proto $scheme; + proxy_connect_timeout 2s; + proxy_read_timeout 10s; + } + } +} diff --git a/test_redis_driver_location_store.py b/test_redis_driver_location_store.py new file mode 100644 index 0000000..24156b5 --- /dev/null +++ b/test_redis_driver_location_store.py @@ -0,0 +1,51 @@ +from datetime import datetime, timezone + +from location_store import RedisDriverLocationStore +from models import DriverLocationEvent + + +class FakeRedis: + def __init__(self): + self.values = {} + + def hset(self, name, key, value): + self.values.setdefault(name, {})[key] = value + + def hget(self, name, key): + return self.values.get(name, {}).get(key) + + def hvals(self, name): + return list(self.values.get(name, {}).values()) + + def hlen(self, name): + return len(self.values.get(name, {})) + + def hdel(self, name, key): + self.values.get(name, {}).pop(key, None) + + def delete(self, name): + self.values.pop(name, None) + + def ping(self): + return True + + +def test_redis_driver_store_round_trip_and_readiness(): + store = RedisDriverLocationStore(client=FakeRedis()) + event = DriverLocationEvent( + driver_id="d1", + lat=40.7128, + lon=-74.0060, + timestamp=datetime.now(timezone.utc), + status="available", + ) + + store.upsert_driver(event) + + assert store.is_ready() + assert store.count() == 1 + assert store.get_driver("d1") == event + assert store.get_all_drivers() == [event] + + store.remove_driver(event.driver_id) + assert store.get_driver(event.driver_id) is None