Multi-Tenancy¶

Complete tenant isolation with per-tenant configurations, data segregation, and resource controls.

Architecture¶

graph TB
    subgraph "Shared Layer"
        A[Load Balancer]
        B[API Gateway]
    end

    subgraph "Tenant A"
        C[Router Instance A]
        D[Config A]
        E[(Database A)]
        F[Cache A]
    end

    subgraph "Tenant B"
        G[Router Instance B]
        H[Config B]
        I[(Database B)]
        J[Cache B]
    end

    subgraph "Tenant C"
        K[Router Instance C]
        L[Config C]
        M[(Database C)]
        N[Cache C]
    end

    A --> B
    B --> C
    B --> G
    B --> K
    C --> D
    C --> E
    C --> F
    G --> H
    G --> I
    G --> J
    K --> L
    K --> M
    K --> N

Tenant Isolation Levels¶

Level 1: Logical Isolation (Standard)¶

Shared infrastructure
Database-level separation
Row-level security
Namespace isolation

Level 2: Process Isolation (Premium)¶

Dedicated process per tenant
Separate caches
Independent configurations
Resource quotas enforced

Level 3: Physical Isolation (Platinum)¶

Dedicated infrastructure
Separate databases
Isolated networks
Custom deployments

Configuration¶

Tenant Creation¶

from stratarouter.enterprise.tenancy import TenantManager

manager = TenantManager()

# Create new tenant
tenant = manager.create_tenant(
    id="acme-corp",
    name="Acme Corporation",
    plan="premium",
    isolation_level="process",
    config={
        "max_routes": 1000,
        "max_requests_per_day": 1000000,
        "budget_limit": 10000.00,
        "data_residency": "us-east-1",
        "retention_days": 90
    }
)

# Configure tenant-specific settings
tenant.configure(
    router_config={
        "dimension": 384,
        "threshold": 0.5,
        "cache_enabled": True
    },
    security={
        "mfa_required": True,
        "ip_allowlist": ["203.0.113.0/24"],
        "session_timeout": 3600
    }
)

Tenant Context¶

from stratarouter.enterprise.tenancy import TenantContext

# Execute in tenant context
with TenantContext(tenant_id="acme-corp"):
    # All operations scoped to this tenant
    router = Router()
    result = router.route(query, embedding)

    # Data automatically isolated
    metrics = get_metrics()  # Only acme-corp's metrics

Data Segregation¶

Database Isolation¶

# Multi-tenant database configuration
database:
  strategy: database_per_tenant

  tenants:
    acme-corp:
      host: db1.example.com
      database: stratarouter_acme
      credentials: /secrets/acme-db

    globex:
      host: db2.example.com
      database: stratarouter_globex
      credentials: /secrets/globex-db

Row-Level Security¶

-- PostgreSQL RLS policies
CREATE POLICY tenant_isolation ON routes
  FOR ALL
  TO authenticated
  USING (tenant_id = current_setting('app.tenant_id')::uuid);

CREATE POLICY tenant_isolation ON executions
  FOR ALL
  TO authenticated
  USING (tenant_id = current_setting('app.tenant_id')::uuid);

Cache Isolation¶

from stratarouter.enterprise.tenancy import TenantCache

cache = TenantCache()

# Keys automatically prefixed with tenant ID
cache.set("routes", routes)  # Actually stores "acme-corp:routes"
cache.get("routes")          # Only returns acme-corp's data

Resource Management¶

Quotas and Limits¶

from stratarouter.enterprise.tenancy import ResourceQuota

quota = ResourceQuota(
    tenant_id="acme-corp",
    limits={
        # Request limits
        "max_requests_per_second": 100,
        "max_requests_per_day": 1000000,

        # Resource limits
        "max_routes": 1000,
        "max_embeddings": 1000000,
        "max_storage_gb": 100,

        # Cost limits
        "daily_budget": 500.00,
        "monthly_budget": 10000.00,

        # Concurrency limits
        "max_concurrent_requests": 50,
        "max_batch_size": 32
    }
)

# Check quota before execution
if quota.check("max_requests_per_second"):
    execute_route()
else:
    raise QuotaExceededException()

Resource Tracking¶

from stratarouter.enterprise.tenancy import ResourceTracker

tracker = ResourceTracker()

# Track resource usage
tracker.record(
    tenant_id="acme-corp",
    resource_type="requests",
    amount=1,
    cost=0.001,
    metadata={"route_id": "billing"}
)

# Get usage report
usage = tracker.get_usage(
    tenant_id="acme-corp",
    period="2026-01",
    group_by=["resource_type", "route_id"]
)

# Usage summary
print(f"Requests: {usage.total_requests}")
print(f"Cost: ${usage.total_cost:.2f}")
print(f"Quota used: {usage.quota_percentage:.1f}%")

Cost Management¶

Chargeback¶

from stratarouter.enterprise.tenancy import Chargeback

chargeback = Chargeback()

# Generate invoice
invoice = chargeback.generate_invoice(
    tenant_id="acme-corp",
    period="2026-01",
    items=[
        {
            "description": "API Requests",
            "quantity": 1000000,
            "unit_price": 0.001,
            "amount": 1000.00
        },
        {
            "description": "Storage",
            "quantity": 50,  # GB
            "unit_price": 0.10,
            "amount": 5.00
        },
        {
            "description": "Support (Premium)",
            "quantity": 1,
            "unit_price": 5000.00,
            "amount": 5000.00
        }
    ],
    total=6005.00
)

# Export invoice
invoice.export("pdf", "invoices/acme-2026-01.pdf")

Showback¶

# Cost allocation without charging
showback = chargeback.generate_showback(
    tenant_id="acme-corp",
    period="2026-01",
    breakdown={
        "by_route": True,
        "by_user": True,
        "by_department": True
    }
)

# Show cost breakdown
print("Route costs:")
for route, cost in showback.by_route.items():
    print(f"  {route}: ${cost:.2f}")

Tenant Management¶

Tenant Lifecycle¶

from stratarouter.enterprise.tenancy import TenantLifecycle

lifecycle = TenantLifecycle()

# Onboard new tenant
await lifecycle.onboard(
    tenant_id="acme-corp",
    steps=[
        "create_database",
        "provision_infrastructure",
        "configure_security",
        "load_initial_data",
        "configure_integrations",
        "verify_setup"
    ]
)

# Suspend tenant (non-payment, security)
await lifecycle.suspend(
    tenant_id="acme-corp",
    reason="payment_overdue",
    preserve_data=True
)

# Resume tenant
await lifecycle.resume(tenant_id="acme-corp")

# Offboard tenant
await lifecycle.offboard(
    tenant_id="acme-corp",
    data_retention_days=30,
    export_data=True
)

Tenant Migration¶

# Migrate tenant to new infrastructure
migration = lifecycle.migrate_tenant(
    tenant_id="acme-corp",
    from_region="us-east-1",
    to_region="eu-west-1",
    strategy="blue_green",
    steps=[
        "replicate_database",
        "sync_cache",
        "update_dns",
        "verify_connectivity",
        "cutover",
        "cleanup_old"
    ]
)

# Monitor migration
status = migration.status()
print(f"Progress: {status.progress}%")

Custom Configurations¶

Per-Tenant Routing¶

# Tenant-specific routing configuration
tenant_configs = {
    "acme-corp": {
        "encoder": "openai-ada-002",
        "threshold": 0.5,
        "fallback_route": "general"
    },
    "globex": {
        "encoder": "custom-medical",
        "threshold": 0.7,
        "fallback_route": "escalation"
    }
}

# Apply tenant config
router = Router(**tenant_configs[tenant_id])

Tenant-Specific Integrations¶

# Configure integrations per tenant
integrations = {
    "acme-corp": {
        "providers": ["openai", "anthropic"],
        "cache_backend": "redis",
        "audit_backend": "splunk"
    },
    "globex": {
        "providers": ["azure-openai", "google"],
        "cache_backend": "memcached",
        "audit_backend": "elasticsearch"
    }
}

Security¶

Tenant Isolation Verification¶

from stratarouter.enterprise.tenancy import IsolationVerifier

verifier = IsolationVerifier()

# Verify tenant isolation
report = verifier.verify(
    tenant_id="acme-corp",
    checks=[
        "database_isolation",
        "cache_isolation",
        "network_isolation",
        "process_isolation",
        "resource_isolation"
    ]
)

if not report.all_passed:
    logger.critical(f"Isolation violation: {report.failures}")
    alert_security_team(report)

Cross-Tenant Access Prevention¶

# Middleware to enforce tenant context
from stratarouter.enterprise.tenancy import TenantMiddleware

@TenantMiddleware.enforce
async def execute_route(route_id: str):
    """Can only access current tenant's routes."""
    route = Route.get(route_id)  # Automatically filtered by tenant
    return await execute(route)

Monitoring¶

Per-Tenant Metrics¶

from stratarouter.enterprise.tenancy import TenantMetrics

metrics = TenantMetrics()

# Record tenant metrics
metrics.record(
    tenant_id="acme-corp",
    metric="requests_total",
    value=1,
    labels={"route": "billing", "status": "success"}
)

# Query tenant metrics
dashboard = metrics.get_dashboard(
    tenant_id="acme-corp",
    period="last_30_days",
    metrics=[
        "requests_total",
        "latency_p99",
        "error_rate",
        "cost_total"
    ]
)

Tenant Health¶

# Monitor tenant health
health = metrics.get_health(tenant_id="acme-corp")

print(f"Status: {health.status}")  # healthy, degraded, critical
print(f"Uptime: {health.uptime}%")
print(f"Error rate: {health.error_rate}%")
print(f"Quota usage: {health.quota_usage}%")

Best Practices¶

Complete Isolation - Use database-per-tenant for sensitive data
Resource Limits - Enforce quotas to prevent noisy neighbors
Monitoring - Track per-tenant metrics and health
Cost Transparency - Provide clear cost breakdowns
Security First - Regular isolation audits
Graceful Degradation - Handle quota exceeded gracefully
Data Residency - Respect data sovereignty requirements

Next Steps¶

Security - Security features
Governance - Policy engine
High Availability - HA setup
Support - Support tiers

Secure multi-tenancy at scale. 🏢