http-reverse-proxy-pool.ae

// Reverse-proxy demo — full enterprise stack.
//
// This example wires every nginx-class feature std.http.proxy ships:
//
//   - Three upstreams with weighted round-robin (3:2:1)
//   - Active health checks (every 5s, 2 OK / 2 fail thresholds)
//   - In-memory LRU response cache (1000 entries, 64 KiB body cap,
//     60s default TTL, Vary-aware key strategy)
//   - Circuit breaker (5 consecutive failures → open 30s)
//   - Per-upstream token-bucket rate limit (200 rps, burst 50)
//   - Idempotent retry on 5xx + transport (3 retries, 100ms base
//     exponential backoff with full jitter, capped at 10s)
//   - W3C Trace-Context injection (generates traceparent when
//     the inbound request didn't supply one)
//   - Active drain on a misbehaving upstream
//   - Prometheus metrics endpoint exposing per-upstream counters
//
// Run:
//
//     ae build examples/stdlib/http-reverse-proxy-pool.ae -o /tmp/proxy
//     /tmp/proxy
//
//     # Three upstreams that you control yourself, e.g.:
//     python3 -m http.server 9001 &
//     python3 -m http.server 9002 &
//     python3 -m http.server 9003 &
//
//     curl -i http://localhost:8080/
//     curl http://localhost:8080/proxy-metrics    # Prometheus surface
//
// See docs/http-reverse-proxy.md for the full reference.

import std.http
import std.http.proxy

const PORT = 8080

handle_metrics(req: ptr, res: ptr, ud: ptr) {
    body = proxy.pool_metrics_text(ud)
    http.response_set_status(res, 200)
    http.response_set_header(res, "Content-Type", "text/plain; version=0.0.4")
    http.response_set_body(res, body)
}

main() {
    server = http.server_create(PORT)
    if server == 0 {
        println("server_create failed")
        return
    }
    http.server_set_keepalive(server, 1, 0, 30000ms)

    // ---- Pool ---------------------------------------------------
    // Weighted round-robin: A serves ~3/6 of traffic, B ~2/6, C ~1/6.
    // 30s request timeout, dial timeout uses platform default,
    // unbounded inflight (set max_inflight_per_up>0 for a per-host
    // concurrency cap).
    pool = proxy.upstream_pool_new("weighted_rr", 30, 0, 0)
    if pool == 0 {
        println("pool_new failed")
        return
    }
    proxy.upstream_add(pool, "http://localhost:9001", 3)
    proxy.upstream_add(pool, "http://localhost:9002", 2)
    proxy.upstream_add(pool, "http://localhost:9003", 1)

    // ---- Health checks -----------------------------------------
    // Probe /health every 5s; 2 OK in a row marks an upstream up,
    // 2 fails in a row marks it down. The health-check thread
    // starts immediately and shuts down when pool_free is called.
    hc_err = proxy.health_checks_enable(pool, "/health", 200, 5000, 2000, 2, 2)
    if hc_err != "" {
        println("health_checks_enable: ${hc_err}")
        return
    }

    // ---- Circuit breaker ---------------------------------------
    // 5 consecutive failures → breaker opens for 30s. After 30s
    // a single half-open probe decides whether to close again.
    proxy.breaker_configure(pool, 5, 30000, 1)

    // ---- Per-upstream rate limit -------------------------------
    // 200 rps cap per upstream with burst capacity of 50. Excess
    // is rejected by the LB picker (request will fall through to
    // another eligible upstream, or 503 if none).
    proxy.rate_limit_set(pool, 200, 50)

    // ---- Cache -------------------------------------------------
    // 1000 entries, 64 KiB max body per entry, 60s default TTL,
    // method+URL+Vary key strategy. The cache observes
    // Cache-Control / Vary / ETag / Last-Modified and performs
    // conditional revalidation on stale hits.
    cache = proxy.cache_new(1000, 65536, 60, "method_url_vary")
    if cache == 0 {
        println("cache_new failed")
        return
    }

    // ---- Per-mount options -------------------------------------
    opts = proxy.opts_new()
    proxy.opts_bind_cache(opts, cache)
    proxy.opts_set_xforwarded(opts, 1, 1, 1) // XFF + XFP + XFH all on
    proxy.opts_set_body_cap(opts, 8 * 1024 * 1024) // 8 MiB
    proxy.opts_set_retry_policy(opts, 3, 100) // 3 retries, 100ms base
    proxy.opts_set_trace_inject(opts, 1) // generate traceparent if absent

    // ---- Optional: drain a host (e.g., during a deploy) --------
    // Uncomment to take A out of rotation without removing it from
    // the pool. In-flight requests finish; new ones skip A.
    // proxy.upstream_drain(pool, "http://localhost:9001")

    // ---- Metrics endpoint --------------------------------------
    // Mounted BEFORE the proxy, on a path the proxy doesn't shadow.
    // Production deployments commonly bind metrics to a separate
    // listener; here we keep one port for clarity.
    http.server_get(server, "/proxy-metrics", handle_metrics, pool)

    // ---- Mount the proxy ---------------------------------------
    // path_prefix "/api" routes /api/* to the upstream pool. Use
    // opts_set_strip_prefix to strip "/api" from the forwarded URL
    // when upstreams expect bare paths.
    err = proxy.mount(server, "/api", pool, opts)
    if err != "" {
        println("mount: ${err}")
        return
    }

    println("listening on http://localhost:${PORT}/api/  (3 upstreams 3:2:1, cache, breaker, retry, rate-limit, traceparent inject)")
    println("prometheus metrics at http://localhost:${PORT}/proxy-metrics")
    start_err = http.server_start(server)
    if start_err != "" {
        println("server_start: ${start_err}")
    }
}