From 602143303bcb467df18369787175438d24a238a6 Mon Sep 17 00:00:00 2001 From: Arnav Sharma <73969466+Arnavsharma2@users.noreply.github.com> Date: Mon, 10 Aug 2026 13:22:44 -0700 Subject: [PATCH] Improve gateway metrics and reliability --- .github/workflows/ci.yml | 40 +++ Makefile | 17 +- README.md | 96 +++++-- deploy/docker/gateway.yaml | 46 ++++ .../docker/grafana/dashboards/gatekeeper.json | 254 +++++++++--------- .../grafana/provisioning/alerting/.gitkeep | 1 + .../grafana/provisioning/plugins/.gitkeep | 1 + deploy/docker/nginx.conf | 6 + deploy/docker/prometheus/prometheus.yml | 3 + deploy/docker/prometheus/rules/gatekeeper.yml | 68 +++++ docker-compose.yml | 1 + internal/cache/cache_test.go | 26 ++ internal/config/config_test.go | 20 ++ internal/observability/metrics.go | 148 ++++++++-- internal/observability/metrics_test.go | 57 ++++ internal/proxy/gateway.go | 88 +++++- internal/proxy/gateway_test.go | 186 +++++++++++++ internal/ratelimit/ratelimit.go | 23 +- internal/ratelimit/ratelimit_test.go | 63 +++++ internal/resilience/circuitbreaker.go | 5 +- internal/resilience/circuitbreaker_test.go | 23 ++ loadtests/baseline.js | 24 +- loadtests/cache.js | 14 +- loadtests/rate_limit.js | 29 +- loadtests/scale.js | 17 +- loadtests/upstream_failure.js | 9 +- 26 files changed, 1044 insertions(+), 221 deletions(-) create mode 100644 .github/workflows/ci.yml create mode 100644 deploy/docker/grafana/provisioning/alerting/.gitkeep create mode 100644 deploy/docker/grafana/provisioning/plugins/.gitkeep create mode 100644 deploy/docker/prometheus/rules/gatekeeper.yml create mode 100644 internal/observability/metrics_test.go diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..21abba6 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,40 @@ +name: CI + +on: + push: + branches: [main] + pull_request: + +permissions: + contents: read + +jobs: + test: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-go@v5 + with: + go-version-file: go.mod + cache: true + + - name: Vet + run: go vet ./... + + - name: Race detector + run: go test -race ./... + + - name: Internal package coverage + shell: bash + run: | + go test ./internal/... -coverprofile=coverage.out + go tool cover -func=coverage.out + coverage_value="$(go tool cover -func=coverage.out | awk '/^total:/ {gsub("%", "", $3); print $3}')" + awk -v coverage="$coverage_value" 'BEGIN { if (coverage < 65) { print "coverage " coverage "% is below 65%"; exit 1 } }' + + - name: Upload coverage profile + uses: actions/upload-artifact@v4 + with: + name: coverage + path: coverage.out diff --git a/Makefile b/Makefile index a4fa277..f7faadd 100644 --- a/Makefile +++ b/Makefile @@ -1,8 +1,18 @@ -.PHONY: test run compose-up compose-down load-baseline load-rate-limit load-cache load-failure +.PHONY: test test-cover test-race benchmark run compose-up compose-down load-baseline load-rate-limit load-cache load-failure load-scale load-all test: go test ./... +test-cover: + go test ./internal/... -coverprofile=coverage.out + go tool cover -func=coverage.out + +test-race: + go test -race ./... + +benchmark: + go test -run '^$$' -bench BenchmarkGatewayUncached -benchmem ./internal/proxy + run: go run ./cmd/gateway -config deploy/docker/gateway.yaml @@ -23,3 +33,8 @@ load-cache: load-failure: k6 run loadtests/upstream_failure.js + +load-scale: + k6 run loadtests/scale.js + +load-all: load-baseline load-cache load-rate-limit load-failure load-scale diff --git a/README.md b/README.md index 5465e7f..48deb52 100644 --- a/README.md +++ b/README.md @@ -6,6 +6,7 @@ ![Prometheus](https://img.shields.io/badge/Prometheus-metrics-E6522C?logo=prometheus&logoColor=white) ![Grafana](https://img.shields.io/badge/Grafana-dashboards-F46800?logo=grafana&logoColor=white) ![k6](https://img.shields.io/badge/k6-load_tested-7D64FF?logo=k6&logoColor=white) +[![CI](https://github.com/Arnavsharma2/Distributed-API-Gateway/actions/workflows/ci.yml/badge.svg)](https://github.com/Arnavsharma2/Distributed-API-Gateway/actions/workflows/ci.yml) A distributed API gateway in Go — adds rate limiting, caching, retries, circuit breaking, and Prometheus observability to backend services across multiple replicas without changing application code. @@ -17,8 +18,9 @@ A distributed API gateway in Go — adds rate limiting, caching, retries, circui | Go backend engineering | Idiomatic Go: interfaces, goroutines, structured JSON logging, YAML config parsing | | Redis patterns | Atomic sliding-window rate limiter implemented as a Lua script to eliminate race conditions under concurrent replicas | | Resilience engineering | Circuit breaker, exponential-backoff retry, and fail-open degradation when Redis is unavailable | -| Observability | 7 Prometheus counters/histograms, provisioned Grafana dashboard, structured per-request logs | -| Performance testing | k6 load tests across 5 scenarios; p95/p99 latency measured at up to 1,000 RPS with active circuit breaking and retries | +| Observability | 16 low-cardinality Prometheus metric families, RED/dependency dashboard, recording rules, and actionable alerts | +| Performance testing | Isolated uncached, cache, rate-limit, failure, and scale workloads with p95/p99, failure-rate, and dropped-iteration gates | +| Quality gates | GitHub Actions runs vet, the race detector, and a 65% internal-package coverage floor (67% current) | | Container orchestration | Multi-service Docker Compose stack: Nginx load balancer, 3 gateway replicas, Redis, Prometheus, Grafana | ## Architecture @@ -43,17 +45,51 @@ flowchart LR ## Benchmark Results -| Scenario | Replicas | RPS | p95 latency | p99 latency | Notes | +The baseline now uses `/api/benchmark/products`, an intentionally uncached and +unlimited route. This isolates proxy overhead instead of accidentally measuring +Redis cache performance. At a fixed 2,000 RPS arrival rate, adding an Nginx +upstream keep-alive pool removed per-request TCP setup and source-port exhaustion: + +| Single-replica baseline | Actual RPS | HTTP failures | p95 latency | p99 latency | Dropped iterations | +|---|---:|---:|---:|---:|---:| +| Before upstream keep-alive | 1,527.12 | 5.90% | 215.82ms | 251.41ms | 13,803 | +| After upstream keep-alive | **1,992.24** | **0.00%** | **0.97ms** | **1.64ms** | **228** | + +That is 30% more completed requests per second, no HTTP failures, and a 99.5% +reduction in p95 latency under the same workload. These are fixed-arrival load +tests, not claims about an absolute hardware ceiling. + +The isolated scenario suite now reports: + +| Scenario | Replicas | Target / actual RPS | p95 | p99 | Scenario signal | |---|---:|---:|---:|---:|---| -| Baseline smoke | 3 | 18.69 | 8.73ms | 18.38ms | Sanity pass | -| Rate-limit smoke | 3 | 200.01 | 6.42ms | 11.75ms | Global sliding-window active | -| Baseline products | 1 | 244.99 | 6.02ms | 46.96ms | Single-instance ceiling | -| Baseline products | 3 | 500.01 | 4.40ms | 8.31ms | **2× throughput, 5× p99 improvement** vs. 1 replica | -| Cache pressure | 3 | 999.94 | 2.20ms | 6.24ms | ~1,000 RPS via Redis cache | -| Rate-limit pressure | 3 | 200.02 | 6.11ms | 12.17ms | Global enforcement across all replicas | -| Upstream failure | 3 | 100.02 | 7.13ms | 13.68ms | Retries and circuit breaking active | +| Uncached proxy | 1 | 2,000 / 1,992.24 | 0.97ms | 1.64ms | 0% HTTP failures | +| Uncached proxy | 3 | 2,000 / 1,981.87 | 1.21ms | 2.73ms | 0% HTTP failures | +| Redis cache | 3 | 1,000 / 996.70 | 1.33ms | 2.92ms | 99.85% hit rate | +| Global rate limit | 3 | 200 / 200.00 | 1.40ms | 1.99ms | Exactly 40 of 6,001 requests admitted | +| Sustained upstream failure | 3 | 100 / 100.02 | 1.52ms | 3.19ms | 99.50% rejected by open circuits | + +The single upstream is shared by all replicas, so the three-replica result is a +resilience check at this arrival rate, not a throughput multiplier. The global +limiter test also verifies that Redis sorted-set members include a random +replica ID; timestamp-plus-local-counter members previously collided across +replicas and admitted 43 requests against a limit of 40. -Scaling from 1 → 3 replicas doubled throughput and cut p99 latency from 46.96ms to 8.31ms on the products route. +Reproduce the current baseline with: + +```bash +docker compose up -d --build --scale gateway=1 +docker run --rm -i \ + -v "$PWD/loadtests:/scripts" \ + grafana/k6:0.54.0 run \ + -e BASE_URL=http://host.docker.internal:8080 \ + -e RATE=2000 /scripts/baseline.js +``` + +Measurements were collected on 2026-08-10 using Docker Desktop and k6 0.54.0. +The test ran for 30 seconds after the stack was warm. Results will vary by host, +so the scripts enforce service-level thresholds and report load-generator drops +separately from HTTP failures. ## Features @@ -64,7 +100,7 @@ Scaling from 1 → 3 replicas doubled throughput and cut p99 latency from 46.96m - Retry policy with exponential backoff for transient upstream failures. - Circuit breaker that opens after repeated failures and recovers after cooldown. - Structured JSON logs with request ID, route, status, cache status, rate-limit status, and latency. -- Prometheus metrics and a provisioned Grafana dashboard. +- Prometheus RED and dependency metrics, recording rules, alerts, and a provisioned Grafana dashboard. - k6 scripts for baseline, rate-limit, cache, failure, and scale experiments. ## Quickstart @@ -93,6 +129,11 @@ curl -s http://localhost:9090/metrics | grep gatekeeper # Unit tests go test ./... +# Race detector, coverage, and microbenchmark +make test-race +make test-cover +make benchmark + # Load tests k6 run loadtests/baseline.js k6 run loadtests/cache.js @@ -143,14 +184,29 @@ If Redis is unavailable, the gateway fails open — rate limiting and cache are | Metric | Description | |---|---| | `gatekeeper_requests_total` | Request count by route and status | -| `gatekeeper_request_duration_seconds` | Latency histogram (p50 / p95 / p99) | -| `gatekeeper_rate_limited_total` | Rate-limited requests by route | +| `gatekeeper_requests_in_flight` | Current concurrency by route and method | +| `gatekeeper_request_duration_seconds` | End-to-end latency by route, method, and status class | +| `gatekeeper_response_size_bytes` | Response-body size distribution | | `gatekeeper_cache_events_total` | Cache hits and misses | -| `gatekeeper_upstream_errors_total` | Upstream error count | -| `gatekeeper_retries_total` | Retry attempts by route | -| `gatekeeper_circuit_state` | Circuit breaker state (0 = closed, 1 = open) | - -The Grafana dashboard visualizes request rate, p95/p99 latency, cache hit rate, upstream errors, and circuit state in real time. +| `gatekeeper_rate_limit_decisions_total` | Allowed, limited, and fail-open decisions | +| `gatekeeper_upstream_requests_total` | Upstream attempts by outcome | +| `gatekeeper_upstream_request_duration_seconds` | Upstream latency by route and outcome | +| `gatekeeper_redis_operations_total` | Redis operations and errors by operation | +| `gatekeeper_redis_operation_duration_seconds` | Redis dependency latency | +| `gatekeeper_retries_total` | Retry attempts by route and reason | +| `gatekeeper_circuit_state` | One-hot closed, open, and half-open state | +| `gatekeeper_circuit_transitions_total` | Circuit state changes | +| `gatekeeper_circuit_rejected_total` | Requests rejected by open circuits | + +The request histogram starts at 0.5ms, so Prometheus can resolve the low-millisecond +percentiles produced by the local benchmark. The Grafana dashboard covers request +rate, 5xx ratio, p50/p95/p99 latency, concurrency, cache effectiveness, upstream +latency, Redis latency, rate-limit decisions, retries, and circuit behavior. Its +route variable supports both focused debugging and fleet-wide views. + +Prometheus also provisions five alerts: target down, high 5xx ratio, high p95 +latency, Redis errors, and an open circuit. Three recording rules precompute the +5-minute request rate, 5xx ratio, and p95 latency used for SLO monitoring. ## Failure Modes @@ -159,4 +215,4 @@ The Grafana dashboard visualizes request rate, p95/p99 latency, cache hit rate, | Redis unavailable | Fails open — rate limiting and cache bypass; requests continue to upstream | | Upstream intermittent 5xx | Retries with exponential backoff; records retry and error metrics per attempt | | Upstream sustained failure | Circuit breaker opens after threshold; returns `503` until cooldown half-open trial | -| Multiple gateway replicas | All replicas share Redis state — limits are enforced globally, not per process | \ No newline at end of file +| Multiple gateway replicas | All replicas share Redis state — limits are enforced globally, not per process | diff --git a/deploy/docker/gateway.yaml b/deploy/docker/gateway.yaml index 45d6b65..a7eb5b1 100644 --- a/deploy/docker/gateway.yaml +++ b/deploy/docker/gateway.yaml @@ -6,6 +6,52 @@ redis: url: redis://redis:6379 routes: + # Dedicated uncached route for measuring proxy overhead. Keeping benchmark + # traffic away from Redis avoids conflating gateway, cache, and rate-limit cost. + - name: benchmark-products + path_prefix: /api/benchmark/products + upstream_url: http://mock-api:3000/products + rate_limit: + enabled: false + cache: + enabled: false + retry: + enabled: false + circuit_breaker: + enabled: false + + # Dedicated cache workload: cache behavior is measured without unrelated + # rate-limit rejections changing the result. + - name: benchmark-cache + path_prefix: /api/benchmark/cache/products + upstream_url: http://mock-api:3000/products + rate_limit: + enabled: false + cache: + enabled: true + ttl_seconds: 30 + retry: + enabled: false + circuit_breaker: + enabled: false + + # Dedicated rate-limit workload backed by a stable upstream. + - name: benchmark-rate-limit + path_prefix: /api/benchmark/rate-limit + upstream_url: http://mock-api:3000/products + rate_limit: + enabled: true + key: header + header: X-User-ID + limit: 40 + window_seconds: 60 + cache: + enabled: false + retry: + enabled: false + circuit_breaker: + enabled: false + - name: products path_prefix: /api/products upstream_url: http://mock-api:3000/products diff --git a/deploy/docker/grafana/dashboards/gatekeeper.json b/deploy/docker/grafana/dashboards/gatekeeper.json index b8e153c..5ff6f85 100644 --- a/deploy/docker/grafana/dashboards/gatekeeper.json +++ b/deploy/docker/grafana/dashboards/gatekeeper.json @@ -1,185 +1,175 @@ { - "annotations": { - "list": [] - }, + "annotations": { "list": [] }, "editable": true, "fiscalYearStartMonth": 0, - "graphTooltip": 0, + "graphTooltip": 1, "id": null, "links": [], "liveNow": false, "panels": [ { - "datasource": { - "type": "prometheus", - "uid": "Prometheus" - }, - "fieldConfig": { - "defaults": { - "unit": "reqps" - }, - "overrides": [] - }, - "gridPos": { - "h": 8, - "w": 8, - "x": 0, - "y": 0 - }, + "datasource": { "type": "prometheus", "uid": "Prometheus" }, + "fieldConfig": { "defaults": { "decimals": 1, "unit": "reqps" }, "overrides": [] }, + "gridPos": { "h": 5, "w": 6, "x": 0, "y": 0 }, "id": 1, - "targets": [ - { - "expr": "sum(rate(gatekeeper_requests_total[1m])) by (route)", - "legendFormat": "{{route}}" - } - ], - "title": "Request Rate by Route", - "type": "timeseries" + "options": { "colorMode": "value", "graphMode": "area", "reduceOptions": { "calcs": ["lastNotNull"], "values": false } }, + "targets": [{ "expr": "sum(rate(gatekeeper_requests_total{route=~\"$route\"}[$__rate_interval]))", "legendFormat": "requests" }], + "title": "Request rate", + "type": "stat" }, { - "datasource": { - "type": "prometheus", - "uid": "Prometheus" - }, + "datasource": { "type": "prometheus", "uid": "Prometheus" }, "fieldConfig": { "defaults": { - "unit": "s" + "decimals": 2, + "thresholds": { "mode": "absolute", "steps": [{ "color": "green", "value": null }, { "color": "yellow", "value": 0.01 }, { "color": "red", "value": 0.05 }] }, + "unit": "percentunit" }, "overrides": [] }, - "gridPos": { - "h": 8, - "w": 8, - "x": 8, - "y": 0 - }, + "gridPos": { "h": 5, "w": 6, "x": 6, "y": 0 }, "id": 2, - "targets": [ - { - "expr": "histogram_quantile(0.95, sum(rate(gatekeeper_request_duration_seconds_bucket[1m])) by (le, route))", - "legendFormat": "p95 {{route}}" - }, - { - "expr": "histogram_quantile(0.99, sum(rate(gatekeeper_request_duration_seconds_bucket[1m])) by (le, route))", - "legendFormat": "p99 {{route}}" - } - ], - "title": "Latency p95 / p99", - "type": "timeseries" + "options": { "colorMode": "value", "graphMode": "area", "reduceOptions": { "calcs": ["lastNotNull"], "values": false } }, + "targets": [{ "expr": "sum(rate(gatekeeper_requests_total{route=~\"$route\",status=~\"5..\"}[$__rate_interval])) / clamp_min(sum(rate(gatekeeper_requests_total{route=~\"$route\"}[$__rate_interval])), 0.001)", "legendFormat": "5xx ratio" }], + "title": "5xx error ratio", + "type": "stat" }, { - "datasource": { - "type": "prometheus", - "uid": "Prometheus" - }, + "datasource": { "type": "prometheus", "uid": "Prometheus" }, "fieldConfig": { "defaults": { - "unit": "short" + "thresholds": { "mode": "absolute", "steps": [{ "color": "green", "value": null }, { "color": "yellow", "value": 0.1 }, { "color": "red", "value": 0.25 }] }, + "unit": "s" }, "overrides": [] }, - "gridPos": { - "h": 8, - "w": 8, - "x": 16, - "y": 0 - }, + "gridPos": { "h": 5, "w": 6, "x": 12, "y": 0 }, "id": 3, + "options": { "colorMode": "value", "graphMode": "area", "reduceOptions": { "calcs": ["lastNotNull"], "values": false } }, + "targets": [{ "expr": "histogram_quantile(0.95, sum(rate(gatekeeper_request_duration_seconds_bucket{route=~\"$route\"}[$__rate_interval])) by (le))", "legendFormat": "p95" }], + "title": "End-to-end p95 latency", + "type": "stat" + }, + { + "datasource": { "type": "prometheus", "uid": "Prometheus" }, + "fieldConfig": { "defaults": { "decimals": 1, "unit": "percentunit" }, "overrides": [] }, + "gridPos": { "h": 5, "w": 6, "x": 18, "y": 0 }, + "id": 4, + "options": { "colorMode": "value", "graphMode": "area", "reduceOptions": { "calcs": ["lastNotNull"], "values": false } }, + "targets": [{ "expr": "sum(rate(gatekeeper_cache_events_total{route=~\"$route\",result=\"hit\"}[$__rate_interval])) / clamp_min(sum(rate(gatekeeper_cache_events_total{route=~\"$route\",result=~\"hit|miss\"}[$__rate_interval])), 0.001)", "legendFormat": "hit ratio" }], + "title": "Cache hit ratio", + "type": "stat" + }, + { + "datasource": { "type": "prometheus", "uid": "Prometheus" }, + "fieldConfig": { "defaults": { "unit": "reqps" }, "overrides": [] }, + "gridPos": { "h": 8, "w": 12, "x": 0, "y": 5 }, + "id": 5, + "targets": [{ "expr": "sum(rate(gatekeeper_requests_total{route=~\"$route\"}[$__rate_interval])) by (route, status)", "legendFormat": "{{route}} {{status}}" }], + "title": "Traffic by route and status", + "type": "timeseries" + }, + { + "datasource": { "type": "prometheus", "uid": "Prometheus" }, + "fieldConfig": { "defaults": { "unit": "s" }, "overrides": [] }, + "gridPos": { "h": 8, "w": 12, "x": 12, "y": 5 }, + "id": 6, "targets": [ - { - "expr": "sum(increase(gatekeeper_rate_limited_total[5m])) by (route)", - "legendFormat": "{{route}}" - } + { "expr": "histogram_quantile(0.50, sum(rate(gatekeeper_request_duration_seconds_bucket{route=~\"$route\"}[$__rate_interval])) by (le, route))", "legendFormat": "p50 {{route}}" }, + { "expr": "histogram_quantile(0.95, sum(rate(gatekeeper_request_duration_seconds_bucket{route=~\"$route\"}[$__rate_interval])) by (le, route))", "legendFormat": "p95 {{route}}" }, + { "expr": "histogram_quantile(0.99, sum(rate(gatekeeper_request_duration_seconds_bucket{route=~\"$route\"}[$__rate_interval])) by (le, route))", "legendFormat": "p99 {{route}}" } ], - "title": "Rate-Limited Requests", + "title": "End-to-end latency percentiles", "type": "timeseries" }, { - "datasource": { - "type": "prometheus", - "uid": "Prometheus" - }, - "fieldConfig": { - "defaults": { - "unit": "percentunit" - }, - "overrides": [] - }, - "gridPos": { - "h": 8, - "w": 8, - "x": 0, - "y": 8 - }, - "id": 4, + "datasource": { "type": "prometheus", "uid": "Prometheus" }, + "fieldConfig": { "defaults": { "unit": "short" }, "overrides": [] }, + "gridPos": { "h": 8, "w": 12, "x": 0, "y": 13 }, + "id": 7, + "targets": [{ "expr": "sum(gatekeeper_requests_in_flight{route=~\"$route\"}) by (route)", "legendFormat": "{{route}}" }], + "title": "Requests in flight", + "type": "timeseries" + }, + { + "datasource": { "type": "prometheus", "uid": "Prometheus" }, + "fieldConfig": { "defaults": { "unit": "s" }, "overrides": [] }, + "gridPos": { "h": 8, "w": 12, "x": 12, "y": 13 }, + "id": 8, "targets": [ - { - "expr": "sum(rate(gatekeeper_cache_events_total{result=\"hit\"}[1m])) by (route) / sum(rate(gatekeeper_cache_events_total{result=~\"hit|miss\"}[1m])) by (route)", - "legendFormat": "{{route}}" - } + { "expr": "histogram_quantile(0.95, sum(rate(gatekeeper_upstream_request_duration_seconds_bucket{route=~\"$route\"}[$__rate_interval])) by (le, route))", "legendFormat": "p95 {{route}}" }, + { "expr": "histogram_quantile(0.99, sum(rate(gatekeeper_upstream_request_duration_seconds_bucket{route=~\"$route\"}[$__rate_interval])) by (le, route))", "legendFormat": "p99 {{route}}" } ], - "title": "Cache Hit Rate", + "title": "Upstream latency percentiles", "type": "timeseries" }, { - "datasource": { - "type": "prometheus", - "uid": "Prometheus" - }, - "gridPos": { - "h": 8, - "w": 8, - "x": 8, - "y": 8 - }, - "id": 5, + "datasource": { "type": "prometheus", "uid": "Prometheus" }, + "fieldConfig": { "defaults": { "unit": "s" }, "overrides": [] }, + "gridPos": { "h": 8, "w": 12, "x": 0, "y": 21 }, + "id": 9, + "targets": [{ "expr": "histogram_quantile(0.95, sum(rate(gatekeeper_redis_operation_duration_seconds_bucket{route=~\"$route\"}[$__rate_interval])) by (le, operation))", "legendFormat": "p95 {{operation}}" }], + "title": "Redis operation p95 latency", + "type": "timeseries" + }, + { + "datasource": { "type": "prometheus", "uid": "Prometheus" }, + "fieldConfig": { "defaults": { "unit": "reqps" }, "overrides": [] }, + "gridPos": { "h": 8, "w": 12, "x": 12, "y": 21 }, + "id": 10, + "targets": [{ "expr": "sum(rate(gatekeeper_rate_limit_decisions_total{route=~\"$route\"}[$__rate_interval])) by (route, result)", "legendFormat": "{{route}} {{result}}" }], + "title": "Rate-limit decisions", + "type": "timeseries" + }, + { + "datasource": { "type": "prometheus", "uid": "Prometheus" }, + "fieldConfig": { "defaults": { "unit": "reqps" }, "overrides": [] }, + "gridPos": { "h": 8, "w": 12, "x": 0, "y": 29 }, + "id": 11, "targets": [ - { - "expr": "sum(rate(gatekeeper_upstream_errors_total[1m])) by (route)", - "legendFormat": "{{route}}" - } + { "expr": "sum(rate(gatekeeper_retries_total{route=~\"$route\"}[$__rate_interval])) by (route, reason)", "legendFormat": "retry {{route}} {{reason}}" }, + { "expr": "sum(rate(gatekeeper_upstream_errors_total{route=~\"$route\"}[$__rate_interval])) by (route)", "legendFormat": "upstream error {{route}}" } ], - "title": "Upstream Errors", + "title": "Retries and upstream errors", "type": "timeseries" }, { - "datasource": { - "type": "prometheus", - "uid": "Prometheus" - }, - "gridPos": { - "h": 8, - "w": 8, - "x": 16, - "y": 8 - }, - "id": 6, + "datasource": { "type": "prometheus", "uid": "Prometheus" }, + "fieldConfig": { "defaults": { "unit": "short" }, "overrides": [] }, + "gridPos": { "h": 8, "w": 12, "x": 12, "y": 29 }, + "id": 12, "targets": [ - { - "expr": "gatekeeper_circuit_state", - "legendFormat": "{{route}} {{state}}" - } + { "expr": "gatekeeper_circuit_state{route=~\"$route\"}", "legendFormat": "{{route}} {{state}}" }, + { "expr": "sum(increase(gatekeeper_circuit_rejected_total{route=~\"$route\"}[5m])) by (route)", "legendFormat": "rejected {{route}}" } ], - "title": "Circuit State", + "title": "Circuit state and rejected requests", "type": "timeseries" } ], "refresh": "5s", "schemaVersion": 39, - "tags": [ - "gatekeeper", - "api-gateway" - ], + "tags": ["gatekeeper", "api-gateway", "slo"], "templating": { - "list": [] - }, - "time": { - "from": "now-15m", - "to": "now" + "list": [ + { + "allValue": ".*", + "current": { "selected": true, "text": "All", "value": "$__all" }, + "datasource": { "type": "prometheus", "uid": "Prometheus" }, + "definition": "label_values(gatekeeper_requests_total, route)", + "includeAll": true, + "label": "Route", + "multi": true, + "name": "route", + "query": { "query": "label_values(gatekeeper_requests_total, route)", "refId": "route-variable" }, + "refresh": 1, + "type": "query" + } + ] }, + "time": { "from": "now-15m", "to": "now" }, "timezone": "browser", - "title": "Gatekeeper API Gateway", + "title": "Gatekeeper API Gateway — RED & Dependencies", "uid": "gatekeeper-api-gateway", - "version": 1, + "version": 2, "weekStart": "" } diff --git a/deploy/docker/grafana/provisioning/alerting/.gitkeep b/deploy/docker/grafana/provisioning/alerting/.gitkeep new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/deploy/docker/grafana/provisioning/alerting/.gitkeep @@ -0,0 +1 @@ + diff --git a/deploy/docker/grafana/provisioning/plugins/.gitkeep b/deploy/docker/grafana/provisioning/plugins/.gitkeep new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/deploy/docker/grafana/provisioning/plugins/.gitkeep @@ -0,0 +1 @@ + diff --git a/deploy/docker/nginx.conf b/deploy/docker/nginx.conf index c4f62b4..9184340 100644 --- a/deploy/docker/nginx.conf +++ b/deploy/docker/nginx.conf @@ -6,6 +6,12 @@ http { upstream gatekeeper_gateway { zone gatekeeper_gateway 64k; server gateway:8080 resolve; + + # Reuse upstream connections. Without an explicit pool, Nginx opens a new + # TCP connection per request and can exhaust ephemeral ports under load. + keepalive 512; + keepalive_requests 10000; + keepalive_timeout 60s; } server { diff --git a/deploy/docker/prometheus/prometheus.yml b/deploy/docker/prometheus/prometheus.yml index 097d76e..21f93a6 100644 --- a/deploy/docker/prometheus/prometheus.yml +++ b/deploy/docker/prometheus/prometheus.yml @@ -1,6 +1,9 @@ global: scrape_interval: 5s +rule_files: + - /etc/prometheus/rules/*.yml + scrape_configs: - job_name: gatekeeper dns_sd_configs: diff --git a/deploy/docker/prometheus/rules/gatekeeper.yml b/deploy/docker/prometheus/rules/gatekeeper.yml new file mode 100644 index 0000000..801cc9b --- /dev/null +++ b/deploy/docker/prometheus/rules/gatekeeper.yml @@ -0,0 +1,68 @@ +groups: + - name: gatekeeper-recording + interval: 15s + rules: + - record: gatekeeper:http_requests:rate5m + expr: sum(rate(gatekeeper_requests_total[5m])) by (route) + + - record: gatekeeper:http_5xx_ratio:rate5m + expr: | + sum(rate(gatekeeper_requests_total{status=~"5.."}[5m])) by (route) + / + clamp_min(sum(rate(gatekeeper_requests_total[5m])) by (route), 0.001) + + - record: gatekeeper:http_request_duration_seconds:p95_5m + expr: | + histogram_quantile( + 0.95, + sum(rate(gatekeeper_request_duration_seconds_bucket[5m])) by (le, route) + ) + + - name: gatekeeper-alerts + rules: + - alert: GatekeeperTargetDown + expr: up{job="gatekeeper"} == 0 + for: 1m + labels: + severity: critical + annotations: + summary: Gatekeeper replica is not being scraped + description: Prometheus has been unable to scrape {{ $labels.instance }} for one minute. + + - alert: GatekeeperHighErrorRate + expr: | + gatekeeper:http_5xx_ratio:rate5m > 0.05 + and on (route) gatekeeper:http_requests:rate5m > 1 + for: 5m + labels: + severity: warning + annotations: + summary: High 5xx rate on {{ $labels.route }} + description: More than 5% of requests have returned 5xx responses for five minutes. + + - alert: GatekeeperHighP95Latency + expr: gatekeeper:http_request_duration_seconds:p95_5m > 0.25 + for: 5m + labels: + severity: warning + annotations: + summary: High p95 latency on {{ $labels.route }} + description: End-to-end p95 latency has exceeded 250 ms for five minutes. + + - alert: GatekeeperRedisErrors + expr: sum(rate(gatekeeper_redis_operations_total{result="error"}[5m])) by (route, operation) > 0.1 + for: 2m + labels: + severity: warning + annotations: + summary: Redis errors on {{ $labels.route }} + description: Redis {{ $labels.operation }} errors are forcing degraded behavior. + + - alert: GatekeeperCircuitOpen + expr: gatekeeper_circuit_state{state="open"} == 1 + for: 1m + labels: + severity: critical + annotations: + summary: Circuit breaker open on {{ $labels.route }} + description: Requests to the route are being rejected while its upstream circuit is open. diff --git a/docker-compose.yml b/docker-compose.yml index a60ee63..b02df06 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -43,6 +43,7 @@ services: - "9090:9090" volumes: - ./deploy/docker/prometheus/prometheus.yml:/etc/prometheus/prometheus.yml:ro + - ./deploy/docker/prometheus/rules:/etc/prometheus/rules:ro depends_on: - gateway diff --git a/internal/cache/cache_test.go b/internal/cache/cache_test.go index 300ab9a..89ca7d3 100644 --- a/internal/cache/cache_test.go +++ b/internal/cache/cache_test.go @@ -17,6 +17,23 @@ func TestKeyVariesByUserID(t *testing.T) { } } +func TestKeyVariesByAuthorizationAndQuery(t *testing.T) { + first := httptest.NewRequest(http.MethodGet, "/api/products?id=1", nil) + first.Header.Set("Authorization", "Bearer first") + second := httptest.NewRequest(http.MethodGet, "/api/products?id=1", nil) + second.Header.Set("Authorization", "Bearer second") + third := httptest.NewRequest(http.MethodGet, "/api/products?id=2", nil) + third.Header.Set("Authorization", "Bearer first") + + firstKey := Key("products", first) + if firstKey == Key("products", second) { + t.Fatal("expected cache key to vary by Authorization") + } + if firstKey == Key("products", third) { + t.Fatal("expected cache key to vary by query string") + } +} + func TestCacheableHeadersDropsHopByHopAndCookies(t *testing.T) { headers := http.Header{} headers.Set("Content-Type", "application/json") @@ -34,3 +51,12 @@ func TestCacheableHeadersDropsHopByHopAndCookies(t *testing.T) { t.Fatal("expected set-cookie header to be dropped") } } + +func TestCacheableHeadersCopiesValues(t *testing.T) { + headers := http.Header{"Vary": {"Accept-Encoding", "Origin"}} + cacheable := CacheableHeaders(headers) + headers["Vary"][0] = "mutated" + if cacheable["Vary"][0] != "Accept-Encoding" { + t.Fatal("expected cacheable headers to own a copy of header values") + } +} diff --git a/internal/config/config_test.go b/internal/config/config_test.go index ac4fc46..1a4c38c 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -65,3 +65,23 @@ func TestValidateRejectsDuplicateRouteNames(t *testing.T) { t.Fatal("expected duplicate route validation error") } } + +func TestValidateRejectsInvalidRequiredFields(t *testing.T) { + tests := []struct { + name string + cfg Config + }{ + {name: "missing redis", cfg: Config{Routes: []RouteConfig{{Name: "route", PathPrefix: "/", UpstreamURL: "http://upstream"}}}}, + {name: "missing routes", cfg: Config{Redis: RedisConfig{URL: "redis://localhost"}}}, + {name: "missing route name", cfg: Config{Redis: RedisConfig{URL: "redis://localhost"}, Routes: []RouteConfig{{PathPrefix: "/", UpstreamURL: "http://upstream"}}}}, + {name: "invalid prefix", cfg: Config{Redis: RedisConfig{URL: "redis://localhost"}, Routes: []RouteConfig{{Name: "route", PathPrefix: "api", UpstreamURL: "http://upstream"}}}}, + {name: "missing upstream", cfg: Config{Redis: RedisConfig{URL: "redis://localhost"}, Routes: []RouteConfig{{Name: "route", PathPrefix: "/api"}}}}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + if err := test.cfg.Validate(); err == nil { + t.Fatal("expected validation error") + } + }) + } +} diff --git a/internal/observability/metrics.go b/internal/observability/metrics.go index afa8633..634c634 100644 --- a/internal/observability/metrics.go +++ b/internal/observability/metrics.go @@ -11,15 +11,29 @@ import ( "github.com/prometheus/client_golang/prometheus/promhttp" ) +var latencyBuckets = []float64{ + 0.0005, 0.001, 0.0025, 0.005, 0.01, 0.025, 0.05, + 0.1, 0.25, 0.5, 1, 2.5, 5, 10, +} + type Metrics struct { - registry *prometheus.Registry - requests *prometheus.CounterVec - latency *prometheus.HistogramVec - cacheEvents *prometheus.CounterVec - rateLimited *prometheus.CounterVec - upstreamErrors *prometheus.CounterVec - retries *prometheus.CounterVec - circuitState *prometheus.GaugeVec + registry *prometheus.Registry + requests *prometheus.CounterVec + requestsInFlight *prometheus.GaugeVec + latency *prometheus.HistogramVec + responseSize *prometheus.HistogramVec + cacheEvents *prometheus.CounterVec + rateLimitDecisions *prometheus.CounterVec + rateLimited *prometheus.CounterVec + upstreamRequests *prometheus.CounterVec + upstreamLatency *prometheus.HistogramVec + upstreamErrors *prometheus.CounterVec + retries *prometheus.CounterVec + redisOperations *prometheus.CounterVec + redisOperationLatency *prometheus.HistogramVec + circuitState *prometheus.GaugeVec + circuitTransitions *prometheus.CounterVec + circuitRejected *prometheus.CounterVec } func NewMetrics() *Metrics { @@ -30,42 +44,90 @@ func NewMetrics() *Metrics { Name: "gatekeeper_requests_total", Help: "Total requests handled by the gateway.", }, []string{"route", "method", "status"}), + requestsInFlight: prometheus.NewGaugeVec(prometheus.GaugeOpts{ + Name: "gatekeeper_requests_in_flight", + Help: "Requests currently being handled by the gateway.", + }, []string{"route", "method"}), latency: prometheus.NewHistogramVec(prometheus.HistogramOpts{ Name: "gatekeeper_request_duration_seconds", - Help: "Gateway request latency in seconds.", - Buckets: []float64{0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2, 5}, - }, []string{"route", "method"}), + Help: "End-to-end gateway request latency in seconds.", + Buckets: latencyBuckets, + }, []string{"route", "method", "status_class"}), + responseSize: prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Name: "gatekeeper_response_size_bytes", + Help: "Gateway response body size in bytes.", + Buckets: []float64{100, 500, 1_000, 5_000, 10_000, 50_000, 100_000, 500_000, 1_000_000, 5_000_000}, + }, []string{"route", "method", "status_class"}), cacheEvents: prometheus.NewCounterVec(prometheus.CounterOpts{ Name: "gatekeeper_cache_events_total", Help: "Cache events by route and result.", }, []string{"route", "result"}), + rateLimitDecisions: prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "gatekeeper_rate_limit_decisions_total", + Help: "Rate-limit decisions by route and result.", + }, []string{"route", "result"}), rateLimited: prometheus.NewCounterVec(prometheus.CounterOpts{ Name: "gatekeeper_rate_limited_total", Help: "Requests rejected by rate limiting.", }, []string{"route"}), + upstreamRequests: prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "gatekeeper_upstream_requests_total", + Help: "Upstream request attempts by route and outcome.", + }, []string{"route", "method", "outcome"}), + upstreamLatency: prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Name: "gatekeeper_upstream_request_duration_seconds", + Help: "Upstream request attempt latency in seconds, including response body reads.", + Buckets: latencyBuckets, + }, []string{"route", "method", "outcome"}), upstreamErrors: prometheus.NewCounterVec(prometheus.CounterOpts{ Name: "gatekeeper_upstream_errors_total", - Help: "Upstream network errors and 5xx responses.", + Help: "Failed upstream attempts, including transport errors and 5xx responses.", }, []string{"route"}), retries: prometheus.NewCounterVec(prometheus.CounterOpts{ Name: "gatekeeper_retries_total", Help: "Retry attempts issued by the gateway.", - }, []string{"route"}), + }, []string{"route", "reason"}), + redisOperations: prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "gatekeeper_redis_operations_total", + Help: "Redis operations issued by the gateway.", + }, []string{"route", "operation", "result"}), + redisOperationLatency: prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Name: "gatekeeper_redis_operation_duration_seconds", + Help: "Redis operation latency in seconds.", + Buckets: latencyBuckets, + }, []string{"route", "operation"}), circuitState: prometheus.NewGaugeVec(prometheus.GaugeOpts{ Name: "gatekeeper_circuit_state", - Help: "Circuit breaker state by route. A value of 1 marks the active state.", + Help: "Circuit breaker state by route. Exactly one state has value 1.", }, []string{"route", "state"}), + circuitTransitions: prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "gatekeeper_circuit_transitions_total", + Help: "Circuit breaker state transitions by route.", + }, []string{"route", "from", "to"}), + circuitRejected: prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "gatekeeper_circuit_rejected_total", + Help: "Requests rejected because a route circuit breaker is open.", + }, []string{"route"}), } registry.MustRegister( collectors.NewGoCollector(), collectors.NewProcessCollector(collectors.ProcessCollectorOpts{}), m.requests, + m.requestsInFlight, m.latency, + m.responseSize, m.cacheEvents, + m.rateLimitDecisions, m.rateLimited, + m.upstreamRequests, + m.upstreamLatency, m.upstreamErrors, m.retries, + m.redisOperations, + m.redisOperationLatency, m.circuitState, + m.circuitTransitions, + m.circuitRejected, ) return m } @@ -74,26 +136,49 @@ func (m *Metrics) Handler() http.Handler { return promhttp.HandlerFor(m.registry, promhttp.HandlerOpts{}) } -func (m *Metrics) RecordRequest(route string, method string, status int, duration time.Duration) { +func (m *Metrics) RequestStarted(route string, method string) { + m.requestsInFlight.WithLabelValues(route, method).Inc() +} + +func (m *Metrics) RecordRequest(route string, method string, status int, responseBytes int, duration time.Duration) { statusLabel := strconv.Itoa(status) + class := StatusClass(status) m.requests.WithLabelValues(route, method, statusLabel).Inc() - m.latency.WithLabelValues(route, method).Observe(duration.Seconds()) + m.latency.WithLabelValues(route, method, class).Observe(duration.Seconds()) + m.responseSize.WithLabelValues(route, method, class).Observe(float64(responseBytes)) + m.requestsInFlight.WithLabelValues(route, method).Dec() } func (m *Metrics) RecordCache(route string, result string) { m.cacheEvents.WithLabelValues(route, result).Inc() } -func (m *Metrics) RecordRateLimited(route string) { - m.rateLimited.WithLabelValues(route).Inc() +func (m *Metrics) RecordRateLimitDecision(route string, result string) { + m.rateLimitDecisions.WithLabelValues(route, result).Inc() + if result == "limited" { + m.rateLimited.WithLabelValues(route).Inc() + } +} + +func (m *Metrics) RecordUpstream(route string, method string, outcome string, duration time.Duration) { + m.upstreamRequests.WithLabelValues(route, method, outcome).Inc() + m.upstreamLatency.WithLabelValues(route, method, outcome).Observe(duration.Seconds()) + if outcome == "transport_error" || outcome == "read_error" || outcome == "5xx" { + m.upstreamErrors.WithLabelValues(route).Inc() + } } -func (m *Metrics) RecordUpstreamError(route string) { - m.upstreamErrors.WithLabelValues(route).Inc() +func (m *Metrics) RecordRetry(route string, reason string) { + m.retries.WithLabelValues(route, reason).Inc() } -func (m *Metrics) RecordRetry(route string) { - m.retries.WithLabelValues(route).Inc() +func (m *Metrics) RecordRedisOperation(route string, operation string, err error, duration time.Duration) { + result := "success" + if err != nil { + result = "error" + } + m.redisOperations.WithLabelValues(route, operation, result).Inc() + m.redisOperationLatency.WithLabelValues(route, operation).Observe(duration.Seconds()) } func (m *Metrics) SetCircuitState(route string, state resilience.State) { @@ -106,3 +191,22 @@ func (m *Metrics) SetCircuitState(route string, state resilience.State) { m.circuitState.WithLabelValues(route, string(candidate)).Set(value) } } + +func (m *Metrics) RecordCircuitTransition(route string, from resilience.State, to resilience.State) { + if from == to { + return + } + m.circuitTransitions.WithLabelValues(route, string(from), string(to)).Inc() + m.SetCircuitState(route, to) +} + +func (m *Metrics) RecordCircuitRejected(route string) { + m.circuitRejected.WithLabelValues(route).Inc() +} + +func StatusClass(status int) string { + if status < 100 || status > 599 { + return "unknown" + } + return strconv.Itoa(status/100) + "xx" +} diff --git a/internal/observability/metrics_test.go b/internal/observability/metrics_test.go new file mode 100644 index 0000000..1c85186 --- /dev/null +++ b/internal/observability/metrics_test.go @@ -0,0 +1,57 @@ +package observability + +import ( + "errors" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/aps/gatekeeper/internal/resilience" +) + +func TestStatusClass(t *testing.T) { + tests := map[int]string{ + 200: "2xx", + 429: "4xx", + 503: "5xx", + 0: "unknown", + } + for status, want := range tests { + if got := StatusClass(status); got != want { + t.Fatalf("StatusClass(%d) = %q, want %q", status, got, want) + } + } +} + +func TestMetricsExposeRequestDependencyAndCircuitSignals(t *testing.T) { + metrics := NewMetrics() + metrics.RequestStarted("products", "GET") + metrics.RecordRequest("products", "GET", 200, 512, 2*time.Millisecond) + metrics.RecordRateLimitDecision("products", "allowed") + metrics.RecordRedisOperation("products", "cache_get", nil, time.Millisecond) + metrics.RecordRedisOperation("products", "cache_set", errors.New("redis unavailable"), 3*time.Millisecond) + metrics.RecordUpstream("products", "GET", "2xx", 1500*time.Microsecond) + metrics.RecordCircuitTransition("products", resilience.StateClosed, resilience.StateOpen) + metrics.RecordCircuitRejected("products") + + recorder := httptest.NewRecorder() + metrics.Handler().ServeHTTP(recorder, httptest.NewRequest("GET", "/metrics", nil)) + body := recorder.Body.String() + + expected := []string{ + `gatekeeper_requests_total{method="GET",route="products",status="200"} 1`, + `gatekeeper_request_duration_seconds_bucket{method="GET",route="products",status_class="2xx",le="0.0025"} 1`, + `gatekeeper_response_size_bytes_count{method="GET",route="products",status_class="2xx"} 1`, + `gatekeeper_requests_in_flight{method="GET",route="products"} 0`, + `gatekeeper_redis_operations_total{operation="cache_set",result="error",route="products"} 1`, + `gatekeeper_upstream_requests_total{method="GET",outcome="2xx",route="products"} 1`, + `gatekeeper_circuit_state{route="products",state="open"} 1`, + `gatekeeper_circuit_rejected_total{route="products"} 1`, + } + for _, metric := range expected { + if !strings.Contains(body, metric) { + t.Errorf("metrics output does not contain %q", metric) + } + } +} diff --git a/internal/proxy/gateway.go b/internal/proxy/gateway.go index 134e86d..ae2f26b 100644 --- a/internal/proxy/gateway.go +++ b/internal/proxy/gateway.go @@ -42,7 +42,7 @@ type upstreamResponse struct { } func NewGateway(cfg *config.Config, redisClient *redis.Client, metrics *observability.Metrics, logger *slog.Logger) *Gateway { - return &Gateway{ + gateway := &Gateway{ cfg: cfg, cache: gatecache.NewRedisCache(redisClient), limiter: ratelimit.NewRedisLimiter(redisClient), @@ -62,15 +62,31 @@ func NewGateway(cfg *config.Config, redisClient *redis.Client, metrics *observab }, }, } + for _, route := range cfg.Routes { + if route.CircuitBreaker.Enabled { + metrics.SetCircuitState(route.Name, resilience.StateClosed) + } + } + return gateway } func (g *Gateway) ServeHTTP(w http.ResponseWriter, req *http.Request) { start := time.Now() requestID := requestID(req) route := MatchRoute(g.cfg.Routes, req.URL.Path) + routeName := "unmatched" + if route != nil { + routeName = route.Name + } + observedWriter := &responseObserver{ResponseWriter: w, status: http.StatusOK} + w = observedWriter + g.metrics.RequestStarted(routeName, req.Method) + defer func() { + g.metrics.RecordRequest(routeName, req.Method, observedWriter.status, observedWriter.bytes, time.Since(start)) + }() + if route == nil { http.Error(w, "no matching route", http.StatusNotFound) - g.metrics.RecordRequest("unmatched", req.Method, http.StatusNotFound, time.Since(start)) return } @@ -80,13 +96,12 @@ func (g *Gateway) ServeHTTP(w http.ResponseWriter, req *http.Request) { upstreamStatus := 0 defer func() { - g.metrics.RecordRequest(route.Name, req.Method, status, time.Since(start)) g.logger.Info("request completed", "request_id", requestID, "route", route.Name, "method", req.Method, "path", req.URL.Path, - "status", status, + "status", observedWriter.status, "upstream_status", upstreamStatus, "duration_ms", time.Since(start).Milliseconds(), "cache", cacheStatus, @@ -96,9 +111,12 @@ func (g *Gateway) ServeHTTP(w http.ResponseWriter, req *http.Request) { rateHeaders := http.Header{} if route.RateLimit.Enabled { + rateLimitStart := time.Now() result, err := g.checkRateLimit(req.Context(), req, *route) + g.metrics.RecordRedisOperation(route.Name, "rate_limit", err, time.Since(rateLimitStart)) if err != nil { rateLimitStatus = "unavailable" + g.metrics.RecordRateLimitDecision(route.Name, "fail_open") g.logger.Warn("rate limiter unavailable; request allowed", "request_id", requestID, "route", route.Name, "error", err) } else { rateLimitStatus = "allowed" @@ -106,20 +124,23 @@ func (g *Gateway) ServeHTTP(w http.ResponseWriter, req *http.Request) { rateHeaders.Set("X-RateLimit-Remaining", strconv.Itoa(result.Remaining)) if !result.Allowed { rateLimitStatus = "limited" + g.metrics.RecordRateLimitDecision(route.Name, "limited") status = http.StatusTooManyRequests rateHeaders.Set("Retry-After", strconv.Itoa(int(result.RetryAfter.Seconds()))) copyHeaders(w.Header(), rateHeaders) w.Header().Set("Content-Type", "application/json") w.WriteHeader(status) _, _ = w.Write([]byte(`{"error":"rate limit exceeded"}`)) - g.metrics.RecordRateLimited(route.Name) return } + g.metrics.RecordRateLimitDecision(route.Name, "allowed") } } if route.Cache.Enabled && req.Method == http.MethodGet { + cacheStart := time.Now() cached, hit, err := g.cache.Get(req.Context(), gatecache.Key(route.Name, req)) + g.metrics.RecordRedisOperation(route.Name, "cache_get", err, time.Since(cacheStart)) if err != nil { cacheStatus = "unavailable" g.metrics.RecordCache(route.Name, "error") @@ -141,15 +162,19 @@ func (g *Gateway) ServeHTTP(w http.ResponseWriter, req *http.Request) { } if breaker, ok := g.breakers.Get(route.Name); ok { + before := breaker.State() if !breaker.Allow() { status = http.StatusServiceUnavailable g.metrics.SetCircuitState(route.Name, breaker.State()) + g.metrics.RecordCircuitRejected(route.Name) copyHeaders(w.Header(), rateHeaders) w.Header().Set("Content-Type", "application/json") w.WriteHeader(status) _, _ = w.Write([]byte(`{"error":"upstream circuit open"}`)) return } + after := breaker.State() + g.metrics.RecordCircuitTransition(route.Name, before, after) } body, err := io.ReadAll(req.Body) @@ -164,7 +189,6 @@ func (g *Gateway) ServeHTTP(w http.ResponseWriter, req *http.Request) { if err != nil { status = http.StatusBadGateway g.recordBreakerFailure(route.Name) - g.metrics.RecordUpstreamError(route.Name) copyHeaders(w.Header(), rateHeaders) w.Header().Set("Content-Type", "application/json") w.WriteHeader(status) @@ -176,7 +200,6 @@ func (g *Gateway) ServeHTTP(w http.ResponseWriter, req *http.Request) { upstreamStatus = resp.status if resp.status >= http.StatusInternalServerError { g.recordBreakerFailure(route.Name) - g.metrics.RecordUpstreamError(route.Name) } else { g.recordBreakerSuccess(route.Name) } @@ -189,7 +212,10 @@ func (g *Gateway) ServeHTTP(w http.ResponseWriter, req *http.Request) { Header: gatecache.CacheableHeaders(resp.header), Body: resp.body, } - if err := g.cache.Set(req.Context(), gatecache.Key(route.Name, req), payload, ttl); err != nil { + cacheStart := time.Now() + err := g.cache.Set(req.Context(), gatecache.Key(route.Name, req), payload, ttl) + g.metrics.RecordRedisOperation(route.Name, "cache_set", err, time.Since(cacheStart)) + if err != nil { cacheStatus = "store_error" g.metrics.RecordCache(route.Name, "store_error") g.logger.Warn("failed to store response in cache", "request_id", requestID, "route", route.Name, "error", err) @@ -256,10 +282,11 @@ func (g *Gateway) forwardWithRetries(ctx context.Context, original *http.Request lastErr = err } - g.metrics.RecordRetry(route.Name) - if err == nil { - g.metrics.RecordUpstreamError(route.Name) + reason := "5xx" + if err != nil { + reason = "transport_error" } + g.metrics.RecordRetry(route.Name, reason) delay := retryDelay(route.Retry.BaseDelayMS, attempt) select { @@ -290,16 +317,20 @@ func (g *Gateway) forwardOnce(ctx context.Context, original *http.Request, route req.Header.Set("X-Request-ID", requestID) appendForwardedFor(req, original) + upstreamStart := time.Now() resp, err := g.client.Do(req) if err != nil { + g.metrics.RecordUpstream(route.Name, original.Method, "transport_error", time.Since(upstreamStart)) return upstreamResponse{}, err } defer resp.Body.Close() respBody, err := io.ReadAll(resp.Body) if err != nil { + g.metrics.RecordUpstream(route.Name, original.Method, "read_error", time.Since(upstreamStart)) return upstreamResponse{}, err } + g.metrics.RecordUpstream(route.Name, original.Method, observability.StatusClass(resp.StatusCode), time.Since(upstreamStart)) return upstreamResponse{ status: resp.StatusCode, header: cloneHeader(resp.Header), @@ -348,16 +379,47 @@ func cacheResponseHeader(status string) string { func (g *Gateway) recordBreakerFailure(routeName string) { if breaker, ok := g.breakers.Get(routeName); ok { + before := breaker.State() breaker.OnFailure() - g.metrics.SetCircuitState(routeName, breaker.State()) + after := breaker.State() + g.metrics.RecordCircuitTransition(routeName, before, after) + g.metrics.SetCircuitState(routeName, after) } } func (g *Gateway) recordBreakerSuccess(routeName string) { if breaker, ok := g.breakers.Get(routeName); ok { + before := breaker.State() breaker.OnSuccess() - g.metrics.SetCircuitState(routeName, breaker.State()) + after := breaker.State() + g.metrics.RecordCircuitTransition(routeName, before, after) + g.metrics.SetCircuitState(routeName, after) + } +} + +type responseObserver struct { + http.ResponseWriter + status int + bytes int + wroteHeader bool +} + +func (w *responseObserver) WriteHeader(status int) { + if w.wroteHeader { + return + } + w.wroteHeader = true + w.status = status + w.ResponseWriter.WriteHeader(status) +} + +func (w *responseObserver) Write(body []byte) (int, error) { + if !w.wroteHeader { + w.WriteHeader(http.StatusOK) } + n, err := w.ResponseWriter.Write(body) + w.bytes += n + return n, err } func requestID(req *http.Request) string { diff --git a/internal/proxy/gateway_test.go b/internal/proxy/gateway_test.go index 5276523..d53f159 100644 --- a/internal/proxy/gateway_test.go +++ b/internal/proxy/gateway_test.go @@ -1,11 +1,19 @@ package proxy import ( + "io" + "log/slog" + "net/http" + "net/http/httptest" "net/url" + "strings" + "sync/atomic" "testing" "time" "github.com/aps/gatekeeper/internal/config" + "github.com/aps/gatekeeper/internal/observability" + "github.com/redis/go-redis/v9" ) func TestMatchRouteChoosesLongestPrefix(t *testing.T) { @@ -54,3 +62,181 @@ func TestRetryDelayExponential(t *testing.T) { t.Fatalf("expected 200ms, got %s", got) } } + +func TestGatewayRecordsActualStatusAndResponseSize(t *testing.T) { + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusCreated) + _, _ = w.Write([]byte("created")) + })) + defer upstream.Close() + + metrics := observability.NewMetrics() + gateway := newTestGateway(upstream.URL, metrics) + recorder := httptest.NewRecorder() + gateway.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/benchmark", nil)) + + if recorder.Code != http.StatusCreated { + t.Fatalf("expected status 201, got %d", recorder.Code) + } + metricsRecorder := httptest.NewRecorder() + metrics.Handler().ServeHTTP(metricsRecorder, httptest.NewRequest(http.MethodGet, "/metrics", nil)) + body := metricsRecorder.Body.String() + if !strings.Contains(body, `gatekeeper_requests_total{method="GET",route="benchmark",status="201"} 1`) { + t.Fatal("expected request counter with the actual response status") + } + if !strings.Contains(body, `gatekeeper_response_size_bytes_sum{method="GET",route="benchmark",status_class="2xx"} 7`) { + t.Fatal("expected response size histogram to observe the seven-byte body") + } +} + +func TestGatewayRecordsUnmatchedRequests(t *testing.T) { + metrics := observability.NewMetrics() + gateway := newTestGateway("http://unused.invalid", metrics) + recorder := httptest.NewRecorder() + gateway.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/missing", nil)) + + if recorder.Code != http.StatusNotFound { + t.Fatalf("expected status 404, got %d", recorder.Code) + } + body := scrapeMetrics(metrics) + if !strings.Contains(body, `gatekeeper_requests_total{method="GET",route="unmatched",status="404"} 1`) { + t.Fatal("expected unmatched request counter") + } +} + +func TestGatewayRetriesUpstream5xx(t *testing.T) { + var attempts atomic.Int32 + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + if attempts.Add(1) == 1 { + http.Error(w, "try again", http.StatusServiceUnavailable) + return + } + _, _ = w.Write([]byte("ok")) + })) + defer upstream.Close() + + metrics := observability.NewMetrics() + route := config.RouteConfig{ + Name: "retry", + PathPrefix: "/retry", + UpstreamURL: upstream.URL, + Retry: config.RetryConfig{ + Enabled: true, + Attempts: 2, + BaseDelayMS: 1, + }, + } + gateway := newTestGatewayForRoute(route, metrics) + recorder := httptest.NewRecorder() + gateway.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/retry", nil)) + + if recorder.Code != http.StatusOK || attempts.Load() != 2 { + t.Fatalf("expected successful second attempt, got status=%d attempts=%d", recorder.Code, attempts.Load()) + } + body := scrapeMetrics(metrics) + for _, expected := range []string{ + `gatekeeper_upstream_requests_total{method="GET",outcome="5xx",route="retry"} 1`, + `gatekeeper_upstream_requests_total{method="GET",outcome="2xx",route="retry"} 1`, + `gatekeeper_retries_total{reason="5xx",route="retry"} 1`, + } { + if !strings.Contains(body, expected) { + t.Errorf("expected metric %q", expected) + } + } +} + +func TestGatewayRejectsWhenCircuitIsOpen(t *testing.T) { + var attempts atomic.Int32 + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + attempts.Add(1) + http.Error(w, "failed", http.StatusInternalServerError) + })) + defer upstream.Close() + + metrics := observability.NewMetrics() + route := config.RouteConfig{ + Name: "failure", + PathPrefix: "/failure", + UpstreamURL: upstream.URL, + CircuitBreaker: config.CircuitBreakerConfig{ + Enabled: true, + FailureThreshold: 1, + CooldownSeconds: 60, + }, + } + gateway := newTestGatewayForRoute(route, metrics) + + first := httptest.NewRecorder() + gateway.ServeHTTP(first, httptest.NewRequest(http.MethodGet, "/failure", nil)) + second := httptest.NewRecorder() + gateway.ServeHTTP(second, httptest.NewRequest(http.MethodGet, "/failure", nil)) + + if first.Code != http.StatusInternalServerError || second.Code != http.StatusServiceUnavailable { + t.Fatalf("expected 500 then 503, got %d then %d", first.Code, second.Code) + } + if attempts.Load() != 1 { + t.Fatalf("expected open circuit to prevent second upstream call, got %d calls", attempts.Load()) + } + body := scrapeMetrics(metrics) + if !strings.Contains(body, `gatekeeper_circuit_transitions_total{from="closed",route="failure",to="open"} 1`) { + t.Fatal("expected closed-to-open transition metric") + } + if !strings.Contains(body, `gatekeeper_circuit_rejected_total{route="failure"} 1`) { + t.Fatal("expected circuit rejection metric") + } +} + +func TestResponseObserverUsesFirstStatusAndCountsBytes(t *testing.T) { + recorder := httptest.NewRecorder() + observer := &responseObserver{ResponseWriter: recorder, status: http.StatusOK} + observer.WriteHeader(http.StatusAccepted) + observer.WriteHeader(http.StatusInternalServerError) + _, _ = observer.Write([]byte("hello")) + + if observer.status != http.StatusAccepted || observer.bytes != 5 { + t.Fatalf("unexpected observation: status=%d bytes=%d", observer.status, observer.bytes) + } +} + +func BenchmarkGatewayUncached(b *testing.B) { + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"ok":true}`)) + })) + defer upstream.Close() + + gateway := newTestGateway(upstream.URL, observability.NewMetrics()) + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + recorder := httptest.NewRecorder() + gateway.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/benchmark", nil)) + if recorder.Code != http.StatusOK { + b.Fatalf("unexpected status %d", recorder.Code) + } + } +} + +func newTestGateway(upstreamURL string, metrics *observability.Metrics) *Gateway { + return newTestGatewayForRoute(config.RouteConfig{ + Name: "benchmark", + PathPrefix: "/benchmark", + UpstreamURL: upstreamURL, + }, metrics) +} + +func newTestGatewayForRoute(route config.RouteConfig, metrics *observability.Metrics) *Gateway { + cfg := &config.Config{ + Redis: config.RedisConfig{URL: "redis://localhost:6379"}, + Routes: []config.RouteConfig{route}, + } + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + redisClient := redis.NewClient(&redis.Options{Addr: "localhost:6379"}) + return NewGateway(cfg, redisClient, metrics, logger) +} + +func scrapeMetrics(metrics *observability.Metrics) string { + recorder := httptest.NewRecorder() + metrics.Handler().ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/metrics", nil)) + return recorder.Body.String() +} diff --git a/internal/ratelimit/ratelimit.go b/internal/ratelimit/ratelimit.go index ae08590..56884e8 100644 --- a/internal/ratelimit/ratelimit.go +++ b/internal/ratelimit/ratelimit.go @@ -2,6 +2,8 @@ package ratelimit import ( "context" + "crypto/rand" + "encoding/hex" "fmt" "net" "net/http" @@ -22,12 +24,13 @@ type Result struct { } type RedisLimiter struct { - client *redis.Client - counter uint64 + client *redis.Client + instanceID string + counter uint64 } func NewRedisLimiter(client *redis.Client) *RedisLimiter { - return &RedisLimiter{client: client} + return &RedisLimiter{client: client, instanceID: newInstanceID()} } var slidingWindowScript = redis.NewScript(` @@ -61,7 +64,7 @@ func (l *RedisLimiter) Allow(ctx context.Context, key string, limit int, window } now := time.Now().UnixMilli() - member := fmt.Sprintf("%d-%d", now, atomic.AddUint64(&l.counter, 1)) + member := l.member(now) values, err := slidingWindowScript.Run(ctx, l.client, []string{key}, now, window.Milliseconds(), limit, member).Result() if err != nil { return Result{}, err @@ -97,6 +100,18 @@ func (l *RedisLimiter) Allow(ctx context.Context, key string, limit int, window }, nil } +func (l *RedisLimiter) member(now int64) string { + return fmt.Sprintf("%s-%d-%d", l.instanceID, now, atomic.AddUint64(&l.counter, 1)) +} + +func newInstanceID() string { + var value [8]byte + if _, err := rand.Read(value[:]); err == nil { + return hex.EncodeToString(value[:]) + } + return strconv.FormatInt(time.Now().UnixNano(), 36) +} + func KeyFromRequest(req *http.Request, route config.RouteConfig) string { identity := "unknown" keyMode := strings.ToLower(route.RateLimit.Key) diff --git a/internal/ratelimit/ratelimit_test.go b/internal/ratelimit/ratelimit_test.go index 17dc385..416288a 100644 --- a/internal/ratelimit/ratelimit_test.go +++ b/internal/ratelimit/ratelimit_test.go @@ -45,6 +45,69 @@ func TestKeyFromRequestUsesHeader(t *testing.T) { } } +func TestKeyFromRequestSupportsNamedHeaderAndMissingIdentity(t *testing.T) { + req := httptest.NewRequest(http.MethodGet, "/api/products", nil) + req.Header.Set("X-API-Key", "api-key-123") + route := config.RouteConfig{ + Name: "products", + RateLimit: config.RateLimitConfig{ + Key: "header:X-API-Key", + }, + } + if got := KeyFromRequest(req, route); got != "gk:rl:products:api-key-123" { + t.Fatalf("unexpected named-header key %q", got) + } + + route.RateLimit.Key = "X-Missing" + if got := KeyFromRequest(req, route); got != "gk:rl:products:missing" { + t.Fatalf("unexpected missing-identity key %q", got) + } +} + +func TestKeyFromRequestUsesRemoteAddress(t *testing.T) { + req := httptest.NewRequest(http.MethodGet, "/api/products", nil) + req.RemoteAddr = "198.51.100.7:4321" + route := config.RouteConfig{Name: "products", RateLimit: config.RateLimitConfig{Key: "ip"}} + if got := KeyFromRequest(req, route); got != "gk:rl:products:198.51.100.7" { + t.Fatalf("unexpected remote-address key %q", got) + } +} + +func TestAsInt64(t *testing.T) { + tests := []struct { + value interface{} + want int64 + }{ + {value: int64(1), want: 1}, + {value: int(2), want: 2}, + {value: "3", want: 3}, + {value: []byte("4"), want: 4}, + } + for _, test := range tests { + got, err := asInt64(test.value) + if err != nil || got != test.want { + t.Fatalf("asInt64(%v) = %d, %v; want %d", test.value, got, err, test.want) + } + } + if _, err := asInt64(true); err == nil { + t.Fatal("expected unsupported type to fail") + } +} + +func TestLimiterMembersAreUniqueAcrossReplicas(t *testing.T) { + first := NewRedisLimiter(nil) + second := NewRedisLimiter(nil) + now := time.Now().UnixMilli() + + firstMember := first.member(now) + if firstMember == first.member(now) { + t.Fatal("expected members on one replica to vary by counter") + } + if firstMember == second.member(now) { + t.Fatal("expected members on different replicas to vary by instance ID") + } +} + func TestRedisLimiterIntegration(t *testing.T) { redisURL := os.Getenv("REDIS_URL") if redisURL == "" { diff --git a/internal/resilience/circuitbreaker.go b/internal/resilience/circuitbreaker.go index 91a1df0..daa5140 100644 --- a/internal/resilience/circuitbreaker.go +++ b/internal/resilience/circuitbreaker.go @@ -64,9 +64,12 @@ func (b *CircuitBreaker) Allow() bool { b.mu.Lock() defer b.mu.Unlock() - if b.state != StateOpen { + if b.state == StateClosed { return true } + if b.state == StateHalfOpen { + return false + } if time.Since(b.openedAt) >= b.cooldown { b.state = StateHalfOpen return true diff --git a/internal/resilience/circuitbreaker_test.go b/internal/resilience/circuitbreaker_test.go index a847a31..dabec53 100644 --- a/internal/resilience/circuitbreaker_test.go +++ b/internal/resilience/circuitbreaker_test.go @@ -35,8 +35,31 @@ func TestCircuitBreakerHalfOpenAfterCooldown(t *testing.T) { if breaker.State() != StateHalfOpen { t.Fatalf("expected half-open state, got %s", breaker.State()) } + if breaker.Allow() { + t.Fatal("only one half-open trial request should be allowed") + } breaker.OnSuccess() if breaker.State() != StateClosed { t.Fatalf("expected closed after successful trial, got %s", breaker.State()) } } + +func TestCircuitBreakerUsesDefaultsAndReopensAfterFailedTrial(t *testing.T) { + breaker := NewCircuitBreaker(0, 0) + for i := 0; i < 5; i++ { + breaker.OnFailure() + } + if breaker.State() != StateOpen { + t.Fatalf("expected default threshold to open circuit, got %s", breaker.State()) + } + + breaker.cooldown = time.Millisecond + time.Sleep(2 * time.Millisecond) + if !breaker.Allow() { + t.Fatal("expected half-open trial") + } + breaker.OnFailure() + if breaker.State() != StateOpen { + t.Fatalf("expected failed trial to reopen circuit, got %s", breaker.State()) + } +} diff --git a/loadtests/baseline.js b/loadtests/baseline.js index 85a2ec4..7663cbb 100644 --- a/loadtests/baseline.js +++ b/loadtests/baseline.js @@ -1,31 +1,35 @@ import http from 'k6/http'; import { check } from 'k6'; +const RATE = Number(__ENV.RATE || 500); +const DURATION = __ENV.DURATION || '30s'; + export const options = { scenarios: { constant_request_rate: { executor: 'constant-arrival-rate', - rate: 500, + rate: RATE, timeUnit: '1s', - duration: '30s', - preAllocatedVUs: 50, - maxVUs: 200, + duration: DURATION, + preAllocatedVUs: Math.max(50, Math.ceil(RATE * 0.05)), + maxVUs: Math.max(200, Math.ceil(RATE * 0.25)), }, }, thresholds: { http_req_failed: ['rate<0.01'], - http_req_duration: ['p(95)<250', 'p(99)<500'], + http_req_duration: ['p(95)<25', 'p(99)<75'], + // A one-percent scheduler allowance avoids treating load-generator startup + // jitter as a service failure while still catching sustained saturation. + dropped_iterations: [`rate<${RATE * 0.01}`], }, + summaryTrendStats: ['avg', 'med', 'p(90)', 'p(95)', 'p(99)', 'max'], }; const BASE_URL = __ENV.BASE_URL || 'http://localhost:8080'; export default function () { - const res = http.get(`${BASE_URL}/api/products`, { - headers: { - 'X-User-ID': `baseline-${__VU}`, - 'X-Forwarded-For': `1.2.3.${__ITER % 250}` - }, + const res = http.get(`${BASE_URL}/api/benchmark/products`, { + tags: { name: 'uncached-products' }, }); check(res, { 'status is 200': (r) => r.status === 200, diff --git a/loadtests/cache.js b/loadtests/cache.js index 9eb5564..970214b 100644 --- a/loadtests/cache.js +++ b/loadtests/cache.js @@ -1,5 +1,8 @@ import http from 'k6/http'; import { check } from 'k6'; +import { Rate } from 'k6/metrics'; + +const cacheHitRate = new Rate('cache_hit_rate'); export const options = { scenarios: { @@ -14,21 +17,26 @@ export const options = { }, thresholds: { http_req_failed: ['rate<0.01'], - http_req_duration: ['p(95)<150', 'p(99)<300'], + http_req_duration: ['p(95)<25', 'p(99)<75'], + cache_hit_rate: ['rate>0.95'], + dropped_iterations: ['rate<10'], }, + summaryTrendStats: ['avg', 'med', 'p(90)', 'p(95)', 'p(99)', 'max'], }; const BASE_URL = __ENV.BASE_URL || 'http://localhost:8080'; export default function () { - const res = http.get(`${BASE_URL}/api/products?cache_key=portfolio-demo`, { + const res = http.get(`${BASE_URL}/api/benchmark/cache/products?cache_key=portfolio-demo`, { headers: { 'X-User-ID': 'cache-demo-user', 'X-Forwarded-For': `10.0.0.${__ITER % 250}` }, }); + const cacheHit = res.headers['X-Cache'] === 'HIT'; + cacheHitRate.add(cacheHit); check(res, { 'status is 200': (r) => r.status === 200, - 'cache header present': (r) => r.headers['X-Cache'] !== undefined, + 'cache outcome reported': (r) => r.headers['X-Cache'] === 'HIT' || r.headers['X-Cache'] === 'MISS', }); } diff --git a/loadtests/rate_limit.js b/loadtests/rate_limit.js index 34d7e39..1909a8e 100644 --- a/loadtests/rate_limit.js +++ b/loadtests/rate_limit.js @@ -1,7 +1,16 @@ import http from 'k6/http'; import { check } from 'k6'; +import { SharedArray } from 'k6/data'; +import { Counter, Rate } from 'k6/metrics'; -http.setResponseCallback(http.expectedStatuses({ min: 200, max: 399 }, 429, 503)); +const rateLimitedRate = new Rate('rate_limited_rate'); +const allowedRequests = new Counter('allowed_requests'); +const generatedKey = new SharedArray('rate-limit-run-key', () => [ + `k6-${Date.now()}-${Math.random()}`, +])[0]; +const RATE_LIMIT_KEY = __ENV.RATE_LIMIT_KEY || generatedKey; + +http.setResponseCallback(http.expectedStatuses({ min: 200, max: 399 }, 429)); export const options = { scenarios: { @@ -15,20 +24,28 @@ export const options = { }, }, thresholds: { - http_req_duration: ['p(95)<300'], - checks: ['rate>0.90'], + http_req_duration: ['p(95)<25', 'p(99)<75'], + checks: ['rate>0.99'], + allowed_requests: ['count==40'], + rate_limited_rate: ['rate>0.99'], + dropped_iterations: ['rate<2'], }, + summaryTrendStats: ['avg', 'med', 'p(90)', 'p(95)', 'p(99)', 'max'], }; const BASE_URL = __ENV.BASE_URL || 'http://localhost:8080'; export default function () { - const res = http.get(`${BASE_URL}/api/flaky`, { - headers: { 'X-User-ID': 'shared-rate-limit-key' }, + const res = http.get(`${BASE_URL}/api/benchmark/rate-limit`, { + headers: { 'X-User-ID': RATE_LIMIT_KEY }, }); + rateLimitedRate.add(res.status === 429); + if (res.status === 200) { + allowedRequests.add(1); + } check(res, { - 'allowed or limited': (r) => r.status === 200 || r.status === 429 || r.status === 503, + 'allowed or limited': (r) => r.status === 200 || r.status === 429, 'rate limit headers exist on 429': (r) => r.status !== 429 || r.headers['X-Ratelimit-Limit'] !== undefined, }); } diff --git a/loadtests/scale.js b/loadtests/scale.js index af07157..3e05c0a 100644 --- a/loadtests/scale.js +++ b/loadtests/scale.js @@ -1,6 +1,8 @@ import http from 'k6/http'; import { check } from 'k6'; +const PEAK_RATE = Number(__ENV.PEAK_RATE || 500); + export const options = { scenarios: { ramping_rate: { @@ -10,26 +12,25 @@ export const options = { preAllocatedVUs: 50, maxVUs: 500, stages: [ - { duration: '15s', target: 200 }, - { duration: '30s', target: 500 }, + { duration: '15s', target: Math.round(PEAK_RATE * 0.4) }, + { duration: '30s', target: PEAK_RATE }, { duration: '15s', target: 0 }, ], }, }, thresholds: { http_req_failed: ['rate<0.02'], - http_req_duration: ['p(95)<300', 'p(99)<600'], + http_req_duration: ['p(95)<50', 'p(99)<100'], + dropped_iterations: [`rate<${PEAK_RATE * 0.01}`], }, + summaryTrendStats: ['avg', 'med', 'p(90)', 'p(95)', 'p(99)', 'max'], }; const BASE_URL = __ENV.BASE_URL || 'http://localhost:8080'; export default function () { - const res = http.get(`${BASE_URL}/api/products?scale=${__ITER % 10}`, { - headers: { - 'X-User-ID': `scale-${__VU}`, - 'X-Forwarded-For': `192.168.1.${__ITER % 250}` - }, + const res = http.get(`${BASE_URL}/api/benchmark/products?scale=${__ITER % 10}`, { + tags: { name: 'uncached-products-scale' }, }); check(res, { 'status is 200': (r) => r.status === 200, diff --git a/loadtests/upstream_failure.js b/loadtests/upstream_failure.js index b441628..16b4a11 100644 --- a/loadtests/upstream_failure.js +++ b/loadtests/upstream_failure.js @@ -1,5 +1,8 @@ import http from 'k6/http'; import { check } from 'k6'; +import { Rate } from 'k6/metrics'; + +const circuitRejectionRate = new Rate('circuit_rejection_rate'); http.setResponseCallback(http.expectedStatuses(500, 503)); @@ -15,15 +18,19 @@ export const options = { }, }, thresholds: { - http_req_duration: ['p(95)<500'], + http_req_duration: ['p(95)<100', 'p(99)<250'], checks: ['rate>0.80'], + circuit_rejection_rate: ['rate>0.90'], + dropped_iterations: ['rate<1'], }, + summaryTrendStats: ['avg', 'med', 'p(90)', 'p(95)', 'p(99)', 'max'], }; const BASE_URL = __ENV.BASE_URL || 'http://localhost:8080'; export default function () { const res = http.get(`${BASE_URL}/api/error`); + circuitRejectionRate.add(res.status === 503); check(res, { 'gateway returns expected degraded statuses': (r) => r.status === 500 || r.status === 503, });