Backpressure Patterns: Staying Alive Under Load

interface AdmissionController {
  shouldAdmit(request: Request): boolean;
}

// Strategy 1: Simple rate limiting
// Allows N requests per second, rejects the rest
class RateLimitAdmission implements AdmissionController {
  private tokens: number;
  private readonly maxTokens: number;
  private readonly refillRate: number;

  shouldAdmit(request: Request): boolean {
    if (this.tokens > 0) {
      this.tokens--;
      return true;
    }
    return false;
  }
}

// Strategy 2: Concurrency limiting
// Allows N requests in-flight simultaneously
class ConcurrencyAdmission implements AdmissionController {
  private inFlight: number = 0;
  private readonly maxConcurrent: number;

  shouldAdmit(request: Request): boolean {
    if (this.inFlight < this.maxConcurrent) {
      this.inFlight++;
      return true;
    }
    return false;
  }
}

// Strategy 3: Latency-based (adaptive)
// Rejects more aggressively as latency increases
class LatencyAdmission implements AdmissionController {
  private recentLatencies: number[] = [];
  private readonly targetLatency: number;

  shouldAdmit(request: Request): boolean {
    const p99 = this.calculateP99();
    const admitProbability = this.targetLatency / Math.max(p99, this.targetLatency);
    return Math.random() < admitProbability;
  }
}

// Strategy 4: Queue depth based
// Rejects when queue exceeds threshold
class QueueDepthAdmission implements AdmissionController {
  private queueDepth: number = 0;
  private readonly maxQueueDepth: number;

  shouldAdmit(request: Request): boolean {
    return this.queueDepth < this.maxQueueDepth;
  }
}

<?php

namespace App\Http\Middleware;

use Closure;
use Illuminate\Http\Request;
use Illuminate\Support\Facades\Cache;
use Illuminate\Support\Facades\Redis;
use Symfony\Component\HttpFoundation\Response;

interface AdmissionController
{
    public function shouldAdmit(Request $request): bool;
}

// Strategy 1: Simple rate limiting
// Allows N requests per second, rejects the rest
class RateLimitAdmission implements AdmissionController
{
    public function __construct(
        private int $maxTokens,
        private float $refillRate,
        private string $key = 'rate_limit'
    ) {}

    public function shouldAdmit(Request $request): bool
    {
        return Redis::throttle($this->key)
            ->allow($this->maxTokens)
            ->every(1 / $this->refillRate)
            ->then(
                fn () => true,
                fn () => false
            );
    }
}

// Strategy 2: Concurrency limiting
// Allows N requests in-flight simultaneously
class ConcurrencyAdmission implements AdmissionController
{
    public function __construct(
        private int $maxConcurrent,
        private string $key = 'concurrency_limit'
    ) {}

    public function shouldAdmit(Request $request): bool
    {
        $inFlight = (int) Redis::get($this->key) ?? 0;

        if ($inFlight < $this->maxConcurrent) {
            Redis::incr($this->key);
            return true;
        }
        return false;
    }

    public function release(): void
    {
        Redis::decr($this->key);
    }
}

// Strategy 3: Latency-based (adaptive)
// Rejects more aggressively as latency increases
class LatencyAdmission implements AdmissionController
{
    public function __construct(
        private float $targetLatencyMs,
        private string $key = 'latencies'
    ) {}

    public function shouldAdmit(Request $request): bool
    {
        $p99 = $this->calculateP99();
        $admitProbability = $this->targetLatencyMs / max($p99, $this->targetLatencyMs);
        return (mt_rand() / mt_getrandmax()) < $admitProbability;
    }

    public function recordLatency(float $latencyMs): void
    {
        Redis::lpush($this->key, $latencyMs);
        Redis::ltrim($this->key, 0, 999); // Keep last 1000 samples
    }

    private function calculateP99(): float
    {
        $latencies = Redis::lrange($this->key, 0, -1);
        if (empty($latencies)) {
            return $this->targetLatencyMs;
        }

        $latencies = array_map('floatval', $latencies);
        sort($latencies);
        $index = (int) (count($latencies) * 0.99);
        return $latencies[min($index, count($latencies) - 1)];
    }
}

// Strategy 4: Queue depth based
// Rejects when queue exceeds threshold
class QueueDepthAdmission implements AdmissionController
{
    public function __construct(
        private int $maxQueueDepth,
        private string $queueName = 'default'
    ) {}

    public function shouldAdmit(Request $request): bool
    {
        $queueDepth = Redis::llen("queues:{$this->queueName}");
        return $queueDepth < $this->maxQueueDepth;
    }
}

// Laravel middleware combining strategies
class AdmissionControlMiddleware
{
    public function __construct(
        private AdmissionController $controller
    ) {}

    public function handle(Request $request, Closure $next): Response
    {
        if (!$this->controller->shouldAdmit($request)) {
            return response()->json([
                'error' => 'service_unavailable',
                'message' => 'Server is at capacity, please retry later',
            ], 503)->withHeaders([
                'Retry-After' => 30,
            ]);
        }

        return $next($request);
    }
}

from abc import ABC, abstractmethod
from dataclasses import dataclass, field
import random
import time
from typing import Any


class AdmissionController(ABC):
    @abstractmethod
    def should_admit(self, request: Any) -> bool:
        pass


# Strategy 1: Simple rate limiting
# Allows N requests per second, rejects the rest
@dataclass
class RateLimitAdmission(AdmissionController):
    max_tokens: int
    refill_rate: float  # tokens per second
    _tokens: float = field(init=False)
    _last_refill: float = field(init=False)

    def __post_init__(self):
        self._tokens = float(self.max_tokens)
        self._last_refill = time.monotonic()

    def should_admit(self, request: Any) -> bool:
        self._refill()
        if self._tokens >= 1:
            self._tokens -= 1
            return True
        return False

    def _refill(self):
        now = time.monotonic()
        elapsed = now - self._last_refill
        self._tokens = min(self.max_tokens, self._tokens + elapsed * self.refill_rate)
        self._last_refill = now


# Strategy 2: Concurrency limiting
# Allows N requests in-flight simultaneously
@dataclass
class ConcurrencyAdmission(AdmissionController):
    max_concurrent: int
    _in_flight: int = field(default=0, init=False)

    def should_admit(self, request: Any) -> bool:
        if self._in_flight < self.max_concurrent:
            self._in_flight += 1
            return True
        return False

    def release(self):
        self._in_flight = max(0, self._in_flight - 1)


# Strategy 3: Latency-based (adaptive)
# Rejects more aggressively as latency increases
@dataclass
class LatencyAdmission(AdmissionController):
    target_latency_ms: float
    _recent_latencies: list[float] = field(default_factory=list, init=False)

    def should_admit(self, request: Any) -> bool:
        p99 = self._calculate_p99()
        admit_probability = self.target_latency_ms / max(p99, self.target_latency_ms)
        return random.random() < admit_probability

    def record_latency(self, latency_ms: float):
        self._recent_latencies.append(latency_ms)
        # Keep only recent samples
        if len(self._recent_latencies) > 1000:
            self._recent_latencies = self._recent_latencies[-1000:]

    def _calculate_p99(self) -> float:
        if not self._recent_latencies:
            return self.target_latency_ms
        sorted_latencies = sorted(self._recent_latencies)
        index = int(len(sorted_latencies) * 0.99)
        return sorted_latencies[min(index, len(sorted_latencies) - 1)]


# Strategy 4: Queue depth based
# Rejects when queue exceeds threshold
@dataclass
class QueueDepthAdmission(AdmissionController):
    max_queue_depth: int
    _queue_depth: int = field(default=0, init=False)

    def should_admit(self, request: Any) -> bool:
        return self._queue_depth < self.max_queue_depth

    def enqueue(self):
        self._queue_depth += 1

    def dequeue(self):
        self._queue_depth = max(0, self._queue_depth - 1)

interface BackoffConfig {
  initialDelayMs: number;
  maxDelayMs: number;
  multiplier: number;
  jitter: number;  // 0-1, percentage of randomization
}

class ExponentialBackoff {
  private attempt: number = 0;

  constructor(private config: BackoffConfig) {}

  getNextDelay(retryAfterHeader?: string): number {
    // Honor server's Retry-After if provided
    if (retryAfterHeader) {
      const serverDelay = parseInt(retryAfterHeader, 10) * 1000;
      if (!isNaN(serverDelay)) {
        return this.addJitter(serverDelay);
      }
    }

    // Calculate exponential backoff
    const exponentialDelay = this.config.initialDelayMs *
      Math.pow(this.config.multiplier, this.attempt);
    const cappedDelay = Math.min(exponentialDelay, this.config.maxDelayMs);

    this.attempt++;
    return this.addJitter(cappedDelay);
  }

  private addJitter(delay: number): number {
    const jitterRange = delay * this.config.jitter;
    const jitter = (Math.random() - 0.5) * 2 * jitterRange;
    return Math.max(0, delay + jitter);
  }

  reset(): void {
    this.attempt = 0;
  }
}

// Utility
const sleep = (ms: number) => new Promise(resolve => setTimeout(resolve, ms));

// Usage
const backoff = new ExponentialBackoff({
  initialDelayMs: 1000,
  maxDelayMs: 60000,
  multiplier: 2,
  jitter: 0.25,
});

async function fetchWithBackoff(url: string, maxRetries: number = 5) {
  for (let i = 0; i < maxRetries; i++) {
    try {
      const response = await fetch(url);

      if (response.status === 429 || response.status === 503) {
        const retryAfter = response.headers.get('Retry-After');
        const delay = backoff.getNextDelay(retryAfter ?? undefined);
        console.log(`Rate limited, waiting ${delay}ms before retry`);
        await sleep(delay);
        continue;
      }

      backoff.reset();
      return response;
    } catch (error) {
      const delay = backoff.getNextDelay();
      console.log(`Request failed, waiting ${delay}ms before retry`);
      await sleep(delay);
    }
  }

  throw new Error(`Failed after ${maxRetries} retries`);
}

import asyncio
import random
from dataclasses import dataclass, field

import httpx


@dataclass
class BackoffConfig:
    initial_delay_ms: float = 1000
    max_delay_ms: float = 60000
    multiplier: float = 2
    jitter: float = 0.25  # 0-1, percentage of randomization


@dataclass
class ExponentialBackoff:
    config: BackoffConfig = field(default_factory=BackoffConfig)
    _attempt: int = field(default=0, init=False)

    def get_next_delay(self, retry_after_header: str | None = None) -> float:
        # Honor server's Retry-After if provided
        if retry_after_header:
            try:
                server_delay = int(retry_after_header) * 1000
                return self._add_jitter(server_delay)
            except ValueError:
                pass

        # Calculate exponential backoff
        exponential_delay = self.config.initial_delay_ms * (
            self.config.multiplier ** self._attempt
        )
        capped_delay = min(exponential_delay, self.config.max_delay_ms)

        self._attempt += 1
        return self._add_jitter(capped_delay)

    def _add_jitter(self, delay: float) -> float:
        jitter_range = delay * self.config.jitter
        jitter = (random.random() - 0.5) * 2 * jitter_range
        return max(0, delay + jitter)

    def reset(self) -> None:
        self._attempt = 0


# Usage
async def fetch_with_backoff(url: str, max_retries: int = 5) -> httpx.Response:
    backoff = ExponentialBackoff()

    async with httpx.AsyncClient() as client:
        for _ in range(max_retries):
            try:
                response = await client.get(url)

                if response.status_code in (429, 503):
                    retry_after = response.headers.get("Retry-After")
                    delay = backoff.get_next_delay(retry_after)
                    print(f"Rate limited, waiting {delay}ms before retry")
                    await asyncio.sleep(delay / 1000)
                    continue

                backoff.reset()
                return response

            except httpx.RequestError:
                delay = backoff.get_next_delay()
                print(f"Request failed, waiting {delay}ms before retry")
                await asyncio.sleep(delay / 1000)

    raise Exception(f"Failed after {max_retries} retries")

// In application code. Gives fine-grained control per operation, not just
// per service. Allows custom fallback logic (return cached data, degraded
// response, etc.). Can protect non-network operations (database queries,
// file I/O). Popular libraries include Opossum (Node), Pybreaker, Gobreaker
enum CircuitState {
  CLOSED = 'closed',
  OPEN = 'open',
  HALF_OPEN = 'half_open',
}

interface CircuitBreakerConfig {
  failureThreshold: number;      // Failures before opening
  successThreshold: number;      // Successes to close from half-open
  timeout: number;               // Time in open state before half-open
}

class CircuitBreaker {
  private state: CircuitState = CircuitState.CLOSED;
  private failures: number = 0;
  private successes: number = 0;
  private lastFailureTime: number = 0;
  private openTime: number = 0;

  constructor(
    private name: string,
    private config: CircuitBreakerConfig,
  ) {}

  async execute<T>(operation: () => Promise<T>, fallback?: () => T): Promise<T> {
    if (this.state === CircuitState.OPEN) {
      if (Date.now() - this.openTime > this.config.timeout) {
        this.state = CircuitState.HALF_OPEN;
        this.successes = 0;
      } else {
        // Circuit is open, fail fast
        if (fallback) return fallback();
        throw new Error(`Circuit ${this.name} is open`);
      }
    }

    try {
      const result = await operation();
      this.onSuccess();
      return result;
    } catch (error) {
      this.onFailure();
      if (fallback) return fallback();
      throw error;
    }
  }

  private onSuccess(): void {
    this.failures = 0;
    if (this.state === CircuitState.HALF_OPEN) {
      this.successes++;
      if (this.successes >= this.config.successThreshold) {
        this.state = CircuitState.CLOSED;
      }
    }
  }

  private onFailure(): void {
    this.failures++;
    this.lastFailureTime = Date.now();

    if (this.failures >= this.config.failureThreshold) {
      this.state = CircuitState.OPEN;
      this.openTime = Date.now();
    }
  }
}

# Istio DestinationRule for Envoy sidecar circuit breaking.
# Pros: Consistent behavior across all services regardless of language,
#       no code changes required — works for legacy services.
# Cons: Coarser granularity (per-service, not per-operation),
#       limited fallback options (can't return cached data).
apiVersion: networking.istio.io/v1beta1
kind: DestinationRule
metadata:
  name: payment-service-circuit-breaker
spec:
  host: payment-service.default.svc.cluster.local
  trafficPolicy:
    connectionPool:
      tcp:
        maxConnections: 100
      http:
        h2UpgradePolicy: UPGRADE
        http1MaxPendingRequests: 100
        http2MaxRequests: 1000
        maxRequestsPerConnection: 10
        maxRetries: 3
    outlierDetection:
      # Trip circuit after 5 consecutive 5xx errors
      consecutive5xxErrors: 5
      # Check every 10 seconds
      interval: 10s
      # Keep circuit open for 30 seconds (Open state)
      baseEjectionTime: 30s
      # Max 50% of hosts can be ejected
      maxEjectionPercent: 50
      # After ejection period, allow probe (Half-Open state)
      # Success returns to Closed; failure extends ejection

# AWS CloudFormation for API Gateway with Lambda integration.
# Pros: Good for protecting external-facing APIs, managed service.
# Cons: Less sophisticated than app-level breakers, no true
#       half-open state, limited to timeout-based protection.
Resources:
  PaymentApi:
    Type: AWS::ApiGateway::RestApi
    Properties:
      Name: payment-api

  PaymentMethod:
    Type: AWS::ApiGateway::Method
    Properties:
      RestApiId: !Ref PaymentApi
      ResourceId: !GetAtt PaymentApi.RootResourceId
      HttpMethod: POST
      AuthorizationType: NONE
      Integration:
        Type: AWS_PROXY
        IntegrationHttpMethod: POST
        Uri: !Sub arn:aws:apigateway:${AWS::Region}:lambda:path/...
        # Timeout acts as implicit circuit breaker
        TimeoutInMillis: 5000
        # Cache responses to reduce backend load
        CacheKeyParameters:
          - method.request.header.X-Request-Id
        CacheNamespace: payments
        # Return cached response when backend fails
        IntegrationResponses:
          - StatusCode: 200
          - StatusCode: 503
            SelectionPattern: '5\d{2}'
            ResponseTemplates:
              application/json: '{"error": "service_unavailable"}'

# Spring Boot application.yml with Resilience4j bulkhead configuration.
# Thread pool bulkheads limit concurrent calls per operation type.
resilience4j:
  bulkhead:
    instances:
      criticalApi:
        maxConcurrentCalls: 50
        maxWaitDuration: 100ms
      search:
        maxConcurrentCalls: 20
        maxWaitDuration: 500ms
      analytics:
        maxConcurrentCalls: 10
        maxWaitDuration: 5s

  thread-pool-bulkhead:
    instances:
      criticalApi:
        maxThreadPoolSize: 50
        coreThreadPoolSize: 25
        queueCapacity: 100
      search:
        maxThreadPoolSize: 20
        coreThreadPoolSize: 10
        queueCapacity: 50
      analytics:
        maxThreadPoolSize: 10
        coreThreadPoolSize: 5
        queueCapacity: 20

# HikariCP connection pools - separate pools per workload
spring:
  datasource:
    primary:
      hikari:
        maximum-pool-size: 100
        minimum-idle: 20
        connection-timeout: 1000
    read-replica:
      hikari:
        maximum-pool-size: 50
        minimum-idle: 10
        connection-timeout: 500

// .NET with Polly's BulkheadPolicy for workload isolation.
// Each bulkhead limits concurrent executions independently.
using Polly;
using Polly.Bulkhead;

public class BulkheadConfiguration
{
    // Critical API: 50 concurrent, 100 queued
    public static AsyncBulkheadPolicy CriticalApiBulkhead { get; } =
        Policy.BulkheadAsync(
            maxParallelization: 50,
            maxQueuingActions: 100,
            onBulkheadRejectedAsync: async context =>
            {
                // Log rejection, increment metrics
                await Task.CompletedTask;
            });

    // Search: 20 concurrent, 50 queued
    public static AsyncBulkheadPolicy SearchBulkhead { get; } =
        Policy.BulkheadAsync(
            maxParallelization: 20,
            maxQueuingActions: 50);

    // Analytics: 10 concurrent, 20 queued (low priority)
    public static AsyncBulkheadPolicy AnalyticsBulkhead { get; } =
        Policy.BulkheadAsync(
            maxParallelization: 10,
            maxQueuingActions: 20);
}

// Usage in a service
public class PaymentService
{
    public async Task<PaymentResult> ProcessPayment(PaymentRequest request)
    {
        return await BulkheadConfiguration.CriticalApiBulkhead
            .ExecuteAsync(async () =>
            {
                // Critical payment processing - isolated from other workloads
                return await _paymentGateway.Process(request);
            });
    }
}

; pgBouncer configuration for connection pool bulkheads.
; Separate pools per database/user isolate workloads at the connection level.

[databases]
; Primary database pool - critical workloads
primary = host=db-primary.example.com port=5432 dbname=app \
          pool_size=100 min_pool_size=20 reserve_pool_size=10

; Read replica pool - search and read-heavy workloads
replica = host=db-replica.example.com port=5432 dbname=app \
          pool_size=50 min_pool_size=10 reserve_pool_size=5

; Analytics database - isolated, lower priority
analytics = host=db-analytics.example.com port=5432 dbname=analytics \
            pool_size=20 min_pool_size=5 reserve_pool_size=2

[pgbouncer]
listen_addr = 0.0.0.0
listen_port = 6432
auth_type = md5
auth_file = /etc/pgbouncer/userlist.txt

; Pool mode: transaction gives best connection reuse
pool_mode = transaction

; Global limits
max_client_conn = 1000
default_pool_size = 20

; Timeout settings - reject rather than queue indefinitely
query_timeout = 30
query_wait_timeout = 5
client_idle_timeout = 300

#!/bin/bash
# Chaos experiment: Verify system survives connection exhaustion

echo "Starting connection exhaustion experiment"

# Hold 90% of database connections
for i in {1..90}; do
  psql -h db.example.com -c "SELECT pg_sleep(300)" &
done

echo "Holding 90 connections for 5 minutes"
echo "Monitoring system behavior..."

# Monitor during experiment
for i in {1..60}; do
  curl -s http://app.example.com/health | jq '.status'
  curl -s http://app.example.com/metrics | grep 'http_requests_total'
  sleep 5
done

# Cleanup
echo "Releasing connections"
pkill -f "pg_sleep"

echo "Experiment complete. Check dashboards for behavior analysis."

#!/usr/bin/env python3
"""Chaos experiment: Verify system survives connection exhaustion."""

import asyncio
import signal
import sys

import asyncpg
import httpx


async def hold_connection(pool: asyncpg.Pool, duration: int) -> None:
    """Acquire a connection and hold it with a long-running query."""
    async with pool.acquire() as conn:
        await conn.execute(f"SELECT pg_sleep({duration})")


async def monitor_health(app_url: str, interval: int, iterations: int) -> None:
    """Poll health and metrics endpoints during the experiment."""
    async with httpx.AsyncClient() as client:
        for _ in range(iterations):
            try:
                health = await client.get(f"{app_url}/health")
                print(f"Health: {health.json().get('status', 'unknown')}")

                metrics = await client.get(f"{app_url}/metrics")
                for line in metrics.text.splitlines():
                    if "http_requests_total" in line:
                        print(line)
            except httpx.RequestError as e:
                print(f"Request failed: {e}")

            await asyncio.sleep(interval)


async def run_experiment() -> None:
    db_url = "postgresql://user:pass@db.example.com/app"
    app_url = "http://app.example.com"
    num_connections = 90
    hold_duration = 300  # 5 minutes

    print("Starting connection exhaustion experiment")

    # Create pool and hold 90% of connections
    pool = await asyncpg.create_pool(db_url, min_size=num_connections, max_size=num_connections)

    print(f"Holding {num_connections} connections for {hold_duration // 60} minutes")
    print("Monitoring system behavior...")

    # Start connection holders and monitoring concurrently
    holders = [hold_connection(pool, hold_duration) for _ in range(num_connections)]
    monitor = monitor_health(app_url, interval=5, iterations=60)

    await asyncio.gather(monitor, *holders, return_exceptions=True)

    # Cleanup
    print("Releasing connections")
    await pool.close()
    print("Experiment complete. Check dashboards for behavior analysis.")


if __name__ == "__main__":
    # Handle Ctrl+C gracefully
    signal.signal(signal.SIGINT, lambda *_: sys.exit(0))
    asyncio.run(run_experiment())

# Pseudocode
metrics:
  # Capacity indicators
  - name: request_rate
    type: counter
    labels: [endpoint, status]
    alert: "rate > capacity_threshold"

  - name: concurrent_requests
    type: gauge
    labels: [service]
    alert: "value > concurrency_limit * 0.9"

  - name: queue_depth
    type: gauge
    labels: [queue_name]
    alert: "value > max_depth * 0.8"

  # Backpressure indicators
  - name: requests_rejected
    type: counter
    labels: [reason]  # rate_limit, queue_full, circuit_open

  - name: shed_requests
    type: counter
    labels: [tier, reason]

  - name: retry_after_seconds
    type: histogram
    labels: [endpoint]

  # Health indicators
  - name: goodput_ratio
    type: gauge
    description: "successful_requests / total_requests"
    alert: "value < 0.9"

  - name: latency_p99
    type: histogram
    labels: [endpoint]
    alert: "value > slo_threshold"

from prometheus_client import Counter, Gauge, Histogram, start_http_server

# Capacity indicators
REQUEST_RATE = Counter(
    'http_requests_total',
    'Total HTTP requests',
    ['endpoint', 'status']
)

CONCURRENT_REQUESTS = Gauge(
    'http_concurrent_requests',
    'Currently in-flight requests',
    ['service']
)

QUEUE_DEPTH = Gauge(
    'queue_depth',
    'Current queue depth',
    ['queue_name']
)

# Backpressure indicators
REQUESTS_REJECTED = Counter(
    'requests_rejected_total',
    'Requests rejected by backpressure mechanisms',
    ['reason']  # rate_limit, queue_full, circuit_open, admission_control
)

SHED_REQUESTS = Counter(
    'requests_shed_total',
    'Requests shed during degradation',
    ['tier', 'reason']
)

RETRY_AFTER = Histogram(
    'retry_after_seconds',
    'Retry-After values returned to clients',
    ['endpoint'],
    buckets=[1, 5, 10, 30, 60, 120, 300]
)

# Health indicators
GOODPUT_RATIO = Gauge(
    'goodput_ratio',
    'Ratio of successful requests to total requests'
)

LATENCY = Histogram(
    'http_request_duration_seconds',
    'Request latency',
    ['endpoint'],
    buckets=[.005, .01, .025, .05, .1, .25, .5, 1, 2.5, 5, 10]
)

# Usage in middleware
def track_request(endpoint: str, status: int, duration: float, rejected_reason: str | None = None):
    REQUEST_RATE.labels(endpoint=endpoint, status=str(status)).inc()
    LATENCY.labels(endpoint=endpoint).observe(duration)

    if rejected_reason:
        REQUESTS_REJECTED.labels(reason=rejected_reason).inc()

from datadog import initialize, statsd

# Initialize Datadog client
initialize(statsd_host='localhost', statsd_port=8125)


class OverloadMetrics:
    """Datadog metrics for overload monitoring."""

    @staticmethod
    def track_request(endpoint: str, status: int, duration: float):
        """Track a completed request."""
        tags = [f'endpoint:{endpoint}', f'status:{status}']

        # Capacity indicator
        statsd.increment('http.requests.total', tags=tags)

        # Health indicator
        statsd.histogram('http.request.duration', duration, tags=tags)

    @staticmethod
    def set_concurrent_requests(service: str, count: int):
        """Update concurrent request gauge."""
        statsd.gauge('http.concurrent_requests', count, tags=[f'service:{service}'])

    @staticmethod
    def set_queue_depth(queue_name: str, depth: int):
        """Update queue depth gauge."""
        statsd.gauge('queue.depth', depth, tags=[f'queue:{queue_name}'])

    @staticmethod
    def track_rejection(reason: str):
        """Track a rejected request."""
        statsd.increment('requests.rejected', tags=[f'reason:{reason}'])

    @staticmethod
    def track_shed(tier: str, reason: str):
        """Track a shed request during degradation."""
        statsd.increment('requests.shed', tags=[f'tier:{tier}', f'reason:{reason}'])

    @staticmethod
    def set_goodput_ratio(ratio: float):
        """Update goodput ratio gauge."""
        statsd.gauge('goodput.ratio', ratio)

    @staticmethod
    def track_retry_after(endpoint: str, seconds: int):
        """Track Retry-After header value."""
        statsd.histogram('retry_after.seconds', seconds, tags=[f'endpoint:{endpoint}'])


# Usage
metrics = OverloadMetrics()
metrics.track_request('/api/orders', 200, 0.045)
metrics.track_rejection('rate_limit')
metrics.set_goodput_ratio(0.92)

import { metrics } from '@opentelemetry/api';

const meter = metrics.getMeter('overload-metrics');

// Capacity indicators
const requestCounter = meter.createCounter('http.requests.total', {
  description: 'Total HTTP requests',
});

const concurrentRequests = meter.createUpDownCounter('http.concurrent_requests', {
  description: 'Currently in-flight requests',
});

const queueDepth = meter.createObservableGauge('queue.depth', {
  description: 'Current queue depth',
});

// Backpressure indicators
const requestsRejected = meter.createCounter('requests.rejected.total', {
  description: 'Requests rejected by backpressure mechanisms',
});

const requestsShed = meter.createCounter('requests.shed.total', {
  description: 'Requests shed during degradation',
});

const retryAfterHistogram = meter.createHistogram('retry_after.seconds', {
  description: 'Retry-After values returned to clients',
  boundaries: [1, 5, 10, 30, 60, 120, 300],
});

// Health indicators
const goodputRatio = meter.createObservableGauge('goodput.ratio', {
  description: 'Ratio of successful requests to total requests',
});

const latencyHistogram = meter.createHistogram('http.request.duration', {
  description: 'Request latency in seconds',
  unit: 's',
  boundaries: [0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10],
});

// Usage in middleware
function trackRequest(
  endpoint: string,
  status: number,
  durationMs: number,
  rejectedReason?: string
) {
  requestCounter.add(1, { endpoint, status: String(status) });
  latencyHistogram.record(durationMs / 1000, { endpoint });

  if (rejectedReason) {
    requestsRejected.add(1, { reason: rejectedReason });
  }
}

function trackRejection(reason: 'rate_limit' | 'queue_full' | 'circuit_open' | 'admission_control') {
  requestsRejected.add(1, { reason });
}

function trackShed(tier: string, reason: string) {
  requestsShed.add(1, { tier, reason });
}