implemented rate limit. add observability

Author: hdser

Dockerfile

@@ -43,7 +43,7 @@ EXPOSE 8000
 
 # Health check
 HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
-    CMD curl -f http://localhost:8000/ || exit 1
+    CMD curl -f http://localhost:8000/health || exit 1
 
 # Run the application
-CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000", "--proxy-headers"]
+CMD ["python", "-m", "app.server"]

README.md

@@ -13,7 +13,8 @@ The service is built with **FastAPI** and features automatic route discovery bas
 - **Routing:** Dynamic — endpoints are auto-generated from the dbt `manifest.json`
 - **Documentation:** OpenAPI (Swagger UI) & ReDoc (auto-generated)
 - **Security:** Header-based API Key authentication (`X-API-Key`)
-- **Rate Limiting:** In-memory throttling per tier (Free/Pro/Unlimited) using `slowapi`
+- **Rate Limiting:** Tier-based per-pod in-memory throttling using `slowapi`
+- **Observability:** Structured JSON logging (stderr), Prometheus metrics (`/metrics`), Grafana dashboard
 
 ---
 
@@ -23,17 +24,29 @@ The service is built with **FastAPI** and features automatic route discovery bas
 /cerebro-api
 ├── Dockerfile               # Multi-stage Docker build definition
 ├── requirements.txt         # Python dependencies
+├── requirements-dev.txt     # Dev/test dependencies (pytest, httpx)
 ├── .env.example             # Template for environment variables
 ├── api_keys.json            # API keys configuration (git-ignored)
 ├── .gitignore               # Git ignore rules
-└── app
-    ├── main.py              # App entry point
-    ├── config.py            # Settings & Env var loading
-    ├── database.py          # ClickHouse client wrapper
-    ├── security.py          # Auth & Rate limiting logic
-    ├── manifest.py          # Logic to download & parse dbt manifest
-    └── factory.py           # ⚙️ The Engine: auto-generates routes
-````
+├── app/
+│   ├── server.py            # Process entrypoint (setup logging, start uvicorn)
+│   ├── main.py              # FastAPI app, middleware, system endpoints
+│   ├── observability.py     # JSON logging, Prometheus metrics, middleware
+│   ├── config.py            # Settings & env var loading
+│   ├── database.py          # ClickHouse client wrapper
+│   ├── security.py          # Auth resolution, access enforcement, rate limiting
+│   ├── manifest.py          # dbt manifest loader with structured logging
+│   ├── router_manager.py    # Dynamic route lifecycle & background refresh
+│   ├── factory.py           # Dynamic route generation engine
+│   └── api_metadata.py      # Endpoint spec parsing & validation
+└── tests/
+    ├── conftest.py           # Shared fixtures (mocked DB, manifest, API keys)
+    ├── test_endpoints.py     # /, /health, /metrics tests
+    ├── test_auth.py          # Auth resolution & tier access tests
+    ├── test_observability.py # JSON formatter, log_event, middleware tests
+    ├── test_rate_limiting.py # Rate limit enforcement & metrics tests
+    └── test_manifest.py      # Manifest refresh & router_manager tests
+```
 
 ---
 
@@ -117,15 +130,30 @@ Create an `api_keys.json` file in your project root:
 ### 5. Run the Server
 
 ```bash
-uvicorn app.main:app --reload
+python -m app.server
+```
+
+For development with auto-reload:
+
+```bash
+uvicorn app.main:app --reload --proxy-headers
 ```
 
 The API will be available at:
 
 * Root: `http://127.0.0.1:8000`
+* Health check: `http://127.0.0.1:8000/health`
+* Prometheus metrics: `http://127.0.0.1:8000/metrics`
 * Interactive Docs (Swagger UI): `http://127.0.0.1:8000/docs`
 * Alternative Docs (ReDoc): `http://127.0.0.1:8000/redoc`
 
+### 6. Run Tests
+
+```bash
+pip install -r requirements-dev.txt
+./venv/bin/pytest tests/
+```
+
 ---
 
 ## API Authentication & Access Tiers
@@ -424,6 +452,54 @@ Create separate models for different time granularities:
 
 ---
 
+## Observability
+
+The API emits structured JSON logs to stderr and exposes Prometheus metrics at `/metrics`.
+
+### Structured Logging
+
+All log output is JSON, one object per line, with fields: `timestamp`, `level`, `logger`, `message`, `event`, plus context-specific fields. Logs never contain raw API keys, SQL text, query parameters, or request bodies.
+
+Key log events:
+
+| Event | Description |
+|-------|-------------|
+| `http_request` | Every HTTP request (method, route, status, duration) |
+| `api_access_denied` | Auth failures (reason, required/provided tier) |
+| `api_rate_limit` | Rate-limit blocks (tier, identity kind) |
+| `clickhouse_query` | ClickHouse queries (category, resource, granularity, tier, row count, duration) |
+| `manifest_refresh` | Manifest reload lifecycle (trigger, status, model count) |
+| `route_install` | Dynamic route registration (path, model, tier, methods) |
+
+### Prometheus Metrics
+
+All metrics are prefixed with `cerebro_api_`. Key metric families:
+
+| Metric | Type | Labels |
+|--------|------|--------|
+| `http_requests_total` | Counter | method, route, status |
+| `http_request_duration_seconds` | Histogram | method, route |
+| `auth_resolutions_total` | Counter | required_tier, result |
+| `access_denied_total` | Counter | required_tier, provided_tier, reason |
+| `rate_limit_decisions_total` | Counter | tier, result, identity_kind |
+| `dynamic_requests_total` | Counter | category, resource, granularity, tier, method, status |
+| `dynamic_request_duration_seconds` | Histogram | category, resource, granularity, tier, method |
+| `clickhouse_query_duration_seconds` | Histogram | category, resource, granularity, tier, status |
+| `clickhouse_query_errors_total` | Counter | category, resource, granularity, tier |
+| `clickhouse_rows_returned` | Histogram | category, resource, granularity, tier |
+| `manifest_refresh_total` | Counter | trigger, status |
+| `manifest_models_loaded` | Gauge | — |
+| `dynamic_routes_registered` | Gauge | — |
+
+### Kubernetes Integration
+
+- **PodMonitor** scrapes `/metrics` on port `http` — Prometheus discovers pods directly
+- `/metrics` is blocked at the public ALB with a fixed-response 403 rule
+- K8s probes use `/health` (returns 200 when ClickHouse is reachable, 503 otherwise)
+- A dedicated Grafana dashboard (`cerebro-api-observability`) provides real-time visibility across traffic, auth, rate limits, dynamic API, ClickHouse, manifest lifecycle, pod resources, and structured logs
+
+---
+
 ## Deployment (Docker)
 
 This service is designed to run as a **stateless container** on Kubernetes.
@@ -532,13 +608,16 @@ curl -X POST http://localhost:8000/v1/system/manifest/refresh \
 
 | File | Purpose |
 |------|---------|
-| `app/main.py` | FastAPI app initialization |
+| `app/server.py` | Process entrypoint — sets up logging before app import |
+| `app/main.py` | FastAPI app, CORS, middleware, system endpoints (`/`, `/health`, `/metrics`) |
+| `app/observability.py` | JSON log formatter, Prometheus metrics, HTTP middleware, observer helpers |
 | `app/config.py` | Settings & environment loading |
 | `app/database.py` | ClickHouse client wrapper |
 | `app/api_metadata.py` | dbt endpoint metadata parsing and validation |
-| `app/security.py` | Authentication & tier access control |
-| `app/manifest.py` | dbt manifest loader |
-| `app/factory.py` | Dynamic route generation engine |
+| `app/security.py` | Auth resolution, access enforcement, rate-limit helpers |
+| `app/manifest.py` | dbt manifest loader with structured logging and model gauge |
+| `app/router_manager.py` | Dynamic route lifecycle, background refresh, manifest refresh metrics |
+| `app/factory.py` | Dynamic route generation engine with instrumentation |
 
 ### Adding Custom Endpoints
 

app/api_metadata.py

@@ -54,6 +54,9 @@ class ApiEndpointSpec:
     sort: List[ApiSortSpec]
     description: str
     metadata_enabled: bool = False
+    category: str = "general"
+    resource: str = "unknown"
+    granularity: str = "none"
 
 
 @dataclass(frozen=True)

app/config.py

@@ -1,18 +1,22 @@
-import os
 import json
-from typing import Dict, List, Optional, Any
-from pydantic_settings import BaseSettings
+import logging
+import os
+from typing import Any, Dict, Optional
+
 from pydantic import field_validator
+from pydantic_settings import BaseSettings
+
+logger = logging.getLogger("cerebro_api.config")
 
 
 def load_api_keys_from_file(filepath: str) -> Dict[str, Any]:
     """Load API keys from a JSON file if it exists."""
     if os.path.exists(filepath):
         try:
-            with open(filepath, 'r') as f:
+            with open(filepath, "r") as f:
                 return json.load(f)
         except Exception as e:
-            print(f"⚠️ Error loading API keys from {filepath}: {e}")
+            logger.warning("Error loading API keys from %s: %s", filepath, e)
     return {}
 
 
@@ -28,15 +32,12 @@ class Settings(BaseSettings):
     API_CONFIG_PATH: str = "./api_config.yaml"
     DBT_MANIFEST_REFRESH_ENABLED: bool = True
     DBT_MANIFEST_REFRESH_INTERVAL_SECONDS: int = 300
-    
+
     # API Keys file path (JSON file with user keys)
     API_KEYS_FILE: str = "./api_keys.json"
 
     # ClickHouse
-    # Option 1: Use URL (for ClickHouse Cloud)
-    CLICKHOUSE_URL: Optional[str] = None  # e.g., "ujt1j3jrk0.eu-central-1.aws.clickhouse.cloud"
-    
-    # Option 2: Individual settings (URL takes precedence if provided)
+    CLICKHOUSE_URL: Optional[str] = None
     CLICKHOUSE_HOST: str = "localhost"
     CLICKHOUSE_PORT: int = 8443
     CLICKHOUSE_USER: str = "default"
@@ -45,75 +46,58 @@ class Settings(BaseSettings):
     CLICKHOUSE_SECURE: bool = True
 
     # Security: API Keys mapped to user info
-    # Can be set via env var OR loaded from API_KEYS_FILE
-    # Format: {
-    #   "sk_live_abc123": {"user": "alice", "tier": "tier0", "org": "Acme Inc"},
-    #   "sk_live_xyz789": {"user": "bob", "tier": "tier2", "org": "Partner Co"},
-    # }
     API_KEYS: Dict[str, Any] = {}
 
-    @field_validator('API_KEYS', mode='before')
+    @field_validator("API_KEYS", mode="before")
     @classmethod
     def normalize_api_keys(cls, v):
-        """
-        Normalize API keys to full user format.
-        Supports both simple format (key -> tier) and full format (key -> {user, tier, org}).
-        """
         if not isinstance(v, dict):
             return {}
-        
+
         normalized = {}
         for key, value in v.items():
             if isinstance(value, str):
-                # Simple format: "sk_key": "tier0" -> convert to full format
                 normalized[key] = {
                     "user": "anonymous",
                     "tier": value,
-                    "org": None
+                    "org": None,
                 }
             elif isinstance(value, dict):
-                # Full format: ensure required fields exist
                 normalized[key] = {
                     "user": value.get("user", "anonymous"),
                     "tier": value.get("tier", "tier0"),
-                    "org": value.get("org")
+                    "org": value.get("org"),
                 }
             else:
-                # Skip invalid entries
                 continue
-        
+
         return normalized
 
-    # Default tier for endpoints without a tier tag (for testing, set to tier0)
     DEFAULT_ENDPOINT_TIER: str = "tier0"
 
-    # Tier definitions with rate limits (requests per minute)
     TIER_RATE_LIMITS: Dict[str, int] = {
-        "tier0": 20,      # Public/Free tier
-        "tier1": 100,     # Partner tier
-        "tier2": 500,     # Premium tier
-        "tier3": 10000,   # Internal/Admin tier
+        "tier0": 20,
+        "tier1": 100,
+        "tier2": 500,
+        "tier3": 10000,
     }
 
     class Config:
         env_file = ".env"
         case_sensitive = True
-        extra = "ignore"  # Allow extra env vars without raising errors
+        extra = "ignore"
 
     def __init__(self, **kwargs):
         super().__init__(**kwargs)
-        # Debug: show if API_KEYS came from env
         if self.API_KEYS:
-            print(f"✅ Loaded {len(self.API_KEYS)} API keys from environment variable")
-        # Load API keys from file if env var is empty and file exists
+            logger.info("Loaded %d API keys from environment variable", len(self.API_KEYS))
         elif self.API_KEYS_FILE:
             file_keys = load_api_keys_from_file(self.API_KEYS_FILE)
             if file_keys:
-                # Normalize the loaded keys
                 self.API_KEYS = self.normalize_api_keys(file_keys)
-                print(f"✅ Loaded {len(self.API_KEYS)} API keys from {self.API_KEYS_FILE}")
+                logger.info("Loaded %d API keys from %s", len(self.API_KEYS), self.API_KEYS_FILE)
             else:
-                print(f"⚠️ No API keys found. Create {self.API_KEYS_FILE} or set API_KEYS env var.")
+                logger.warning("No API keys found. Create %s or set API_KEYS env var.", self.API_KEYS_FILE)
 
 
 settings = Settings()

app/database.py

@@ -1,37 +1,36 @@
+import logging
+from typing import Any, Dict, List
+
 import clickhouse_connect
+
 from app.config import settings
-from typing import Any, List, Dict
+
+logger = logging.getLogger("cerebro_api.database")
+
 
 class ClickHouseClient:
     _client = None
 
     @classmethod
     def get_client(cls):
-        """
-        Returns a singleton ClickHouse client instance.
-        """
         if cls._client is None:
             cls._client = clickhouse_connect.get_client(
                 host=settings.CLICKHOUSE_URL,
                 port=settings.CLICKHOUSE_PORT,
                 username=settings.CLICKHOUSE_USER,
                 password=settings.CLICKHOUSE_PASSWORD,
                 database=settings.CLICKHOUSE_DATABASE,
-                secure=settings.CLICKHOUSE_SECURE
+                secure=settings.CLICKHOUSE_SECURE,
             )
         return cls._client
 
     @classmethod
     def query(cls, query_str: str, parameters: Dict[str, Any] = None) -> List[Dict]:
-        """
-        Executes a parameterized query and returns a list of dictionaries.
-        """
         client = cls.get_client()
         try:
             result = client.query(query_str, parameters=parameters)
             columns = result.column_names
             return [dict(zip(columns, row)) for row in result.result_rows]
         except Exception as e:
-            # In a real app, you might want to log this to Sentry/Datadog
-            print(f"DB Error: {e}")
-            raise e
+            logger.error("ClickHouse query error: %s", e)
+            raise e