diff --git a/apisix/plugins/api-breaker.lua b/apisix/plugins/api-breaker.lua index 440f0ee60940..9d7600738564 100644 --- a/apisix/plugins/api-breaker.lua +++ b/apisix/plugins/api-breaker.lua @@ -28,6 +28,17 @@ if not shared_buffer then error("failed to get ngx.shared dict when load plugin " .. plugin_name) end +-- Circuit breaker states (only for ratio policy) +local CLOSED = 0 +local OPEN = 1 +local HALF_OPEN = 2 + +-- Number of fixed time buckets used to approximate the sliding window for +-- the ratio policy. Each bucket is keyed by its own epoch, so aging out old +-- data is just a matter of the epoch falling outside the window -- no +-- request ever has to "reset" shared state that another request is +-- concurrently reading or writing. +local NUM_BUCKETS = 10 local schema = { type = "object", @@ -61,70 +72,227 @@ local schema = { type = "integer", minimum = 3, default = 300, + description = "Circuit breaker duration in seconds " .. + "(applies to both count and ratio policies)" }, - unhealthy = { - type = "object", - properties = { - http_statuses = { - type = "array", - minItems = 1, - items = { + policy = { + type = "string", + enum = { "unhealthy-count", "unhealthy-ratio" }, + default = "unhealthy-count", + } + }, + required = { "break_response_code" }, + ["if"] = { + properties = { + policy = { + enum = { "unhealthy-count" }, + }, + }, + }, + ["then"] = { + properties = { + unhealthy = { + type = "object", + properties = { + http_statuses = { + type = "array", + minItems = 1, + items = { + type = "integer", + minimum = 500, + maximum = 599, + }, + uniqueItems = true, + default = { 500 } + }, + failures = { type = "integer", - minimum = 500, - maximum = 599, + minimum = 1, + default = 3, + } + }, + default = { http_statuses = { 500 }, failures = 3 } + }, + healthy = { + type = "object", + properties = { + http_statuses = { + type = "array", + minItems = 1, + items = { + type = "integer", + minimum = 200, + maximum = 499, + }, + uniqueItems = true, + default = { 200 } }, - uniqueItems = true, - default = {500} + successes = { + type = "integer", + minimum = 1, + default = 3, + } + }, + default = { http_statuses = { 200 }, successes = 3 } + } + } + }, + ["else"] = { + ["if"] = { + properties = { + policy = { + enum = { "unhealthy-ratio" }, }, - failures = { - type = "integer", - minimum = 1, - default = 3, - } }, - default = {http_statuses = {500}, failures = 3} }, - healthy = { - type = "object", + ["then"] = { properties = { - http_statuses = { - type = "array", - minItems = 1, - items = { - type = "integer", - minimum = 200, - maximum = 499, + unhealthy = { + type = "object", + properties = { + http_statuses = { + type = "array", + minItems = 1, + items = { + type = "integer", + minimum = 500, + maximum = 599, + }, + uniqueItems = true, + default = { 500 } + }, + error_ratio = { + type = "number", + minimum = 0, + maximum = 1, + default = 0.5, + description = "Failure rate threshold to trigger circuit breaker" + }, + min_request_threshold = { + type = "integer", + minimum = 1, + default = 10, + description = "Minimum number of calls before " .. + "circuit breaker can be triggered" + }, + sliding_window_size = { + type = "integer", + minimum = 10, + maximum = 3600, + default = 300, + description = "Size of the sliding window in seconds" + }, + half_open_max_calls = { + type = "integer", + minimum = 1, + maximum = 20, + default = 3, + description = "Number of permitted calls when " .. + "circuit breaker is half-open" + } }, - uniqueItems = true, - default = {200} + default = { + http_statuses = { 500 }, + error_ratio = 0.5, + min_request_threshold = 10, + sliding_window_size = 300, + half_open_max_calls = 3 + } }, - successes = { - type = "integer", - minimum = 1, - default = 3, + healthy = { + type = "object", + properties = { + http_statuses = { + type = "array", + minItems = 1, + items = { + type = "integer", + minimum = 200, + maximum = 499, + }, + uniqueItems = true, + default = { 200 } + }, + success_ratio = { + type = "number", + minimum = 0, + maximum = 1, + default = 0.6, + description = "Success rate threshold to close circuit breaker " .. + "from half-open state" + } + }, + default = { http_statuses = { 200 }, success_ratio = 0.6 } } - }, - default = {http_statuses = {200}, successes = 3} + } } - }, - required = {"break_response_code"}, + } } - +-- Key generation functions (based on latest APISIX version) local function gen_healthy_key(ctx) return "healthy-" .. core.request.get_host(ctx) .. ctx.var.uri end - local function gen_unhealthy_key(ctx) return "unhealthy-" .. core.request.get_host(ctx) .. ctx.var.uri end - local function gen_lasttime_key(ctx) return "unhealthy-lasttime" .. core.request.get_host(ctx) .. ctx.var.uri end +-- Ratio-policy state is scoped to the plugin's configuration identity +-- (route/service + config version), not the request URI. Keying by URI +-- would let a high-cardinality or wildcard route create unbounded entries +-- in the shared dict; keying by conf_version also means updating the +-- plugin's config starts the breaker fresh instead of inheriting counters +-- collected under a different configuration (same approach as the +-- limit-count and limit-conn plugins). +local function gen_breaker_id(ctx) + return (ctx.conf_type or "") .. "#" .. (ctx.conf_id or "") .. "#" .. + (ctx.conf_version or "") +end + +local function gen_state_key(id) + return "cb-state-" .. id +end + +local function gen_last_state_change_key(id) + return "cb-last-change-" .. id +end + +local function gen_half_open_calls_key(id) + return "cb-half-open-calls-" .. id +end + +local function gen_half_open_success_key(id) + return "cb-half-open-success-" .. id +end + +local function gen_half_open_completed_key(id) + return "cb-half-open-completed-" .. id +end + +local function gen_transition_lock_key(id) + return "cb-transition-" .. id +end + +local function get_bucket_size(window_size) + local size = math.floor((window_size or 300) / NUM_BUCKETS) + if size < 1 then + size = 1 + end + return size +end + +local function gen_bucket_req_key(id, bucket_size, epoch) + return "cb-breq-" .. id .. "-" .. bucket_size .. "-" .. epoch +end + +local function gen_bucket_fail_key(id, bucket_size, epoch) + return "cb-bfail-" .. id .. "-" .. bucket_size .. "-" .. epoch +end local _M = { version = 0.1, @@ -133,13 +301,129 @@ local _M = { schema = schema, } - function _M.check_schema(conf) return core.schema.check(schema, conf) end +-- State/half-open bookkeeping needs to outlive the OPEN wait plus the +-- half-open evaluation, but must still be bounded so idle routes don't hold +-- shared-dict entries forever. +local function get_state_ttl(conf) + return (conf.max_breaker_sec or 300) * 4 +end -function _M.access(conf, ctx) +-- Bucket entries must remain readable for the full span of buckets a +-- sliding-window sum can touch, plus a little slack. +local function get_bucket_ttl(bucket_size) + return bucket_size * (NUM_BUCKETS + 1) +end + +-- Circuit breaker state management functions +local function get_circuit_breaker_state(id) + local state, err = shared_buffer:get(gen_state_key(id)) + if err then + core.log.warn("failed to get circuit breaker state: ", err) + return CLOSED + end + return state or CLOSED +end + +local function set_circuit_breaker_state(conf, id, state) + local ttl = get_state_ttl(conf) + local current_time = ngx.time() + + shared_buffer:set(gen_state_key(id), state, ttl) + shared_buffer:set(gen_last_state_change_key(id), current_time, ttl) + + core.log.info("api-breaker: state changed to ", state, " for ", id, " at ", current_time) +end + +-- Sliding window bookkeeping (ratio policy). Each bucket is named after the +-- epoch it belongs to, so it never needs to be raced-over by concurrent +-- requests deciding whether to reset it -- a bucket for an epoch outside the +-- window simply isn't summed, and eventually expires via its own TTL. +local function record_window_sample(conf, id, is_failure) + local window_size = conf.unhealthy.sliding_window_size or 300 + local bucket_size = get_bucket_size(window_size) + local ttl = get_bucket_ttl(bucket_size) + local epoch = math.floor(ngx.time() / bucket_size) + + local _, err = shared_buffer:incr(gen_bucket_req_key(id, bucket_size, epoch), 1, 0, ttl) + if err then + core.log.warn("failed to incr window request bucket: ", err) + end + + if is_failure then + local _, ferr = shared_buffer:incr( + gen_bucket_fail_key(id, bucket_size, epoch), 1, 0, ttl) + if ferr then + core.log.warn("failed to incr window failure bucket: ", ferr) + end + end +end + +local function sum_window(conf, id) + local window_size = conf.unhealthy.sliding_window_size or 300 + local bucket_size = get_bucket_size(window_size) + local current_epoch = math.floor(ngx.time() / bucket_size) + + local total_requests = 0 + local total_failures = 0 + for i = 0, NUM_BUCKETS - 1 do + local epoch = current_epoch - i + total_requests = total_requests + + (shared_buffer:get(gen_bucket_req_key(id, bucket_size, epoch)) or 0) + total_failures = total_failures + + (shared_buffer:get(gen_bucket_fail_key(id, bucket_size, epoch)) or 0) + end + + return total_requests, total_failures +end + +local function clear_window(conf, id) + local window_size = conf.unhealthy.sliding_window_size or 300 + local bucket_size = get_bucket_size(window_size) + local current_epoch = math.floor(ngx.time() / bucket_size) + + for i = 0, NUM_BUCKETS - 1 do + local epoch = current_epoch - i + shared_buffer:delete(gen_bucket_req_key(id, bucket_size, epoch)) + shared_buffer:delete(gen_bucket_fail_key(id, bucket_size, epoch)) + end +end + +local function open_breaker(conf, id) + set_circuit_breaker_state(conf, id, OPEN) + shared_buffer:delete(gen_half_open_calls_key(id)) + shared_buffer:delete(gen_half_open_success_key(id)) + shared_buffer:delete(gen_half_open_completed_key(id)) +end + +local function close_breaker(conf, id) + shared_buffer:delete(gen_state_key(id)) + shared_buffer:delete(gen_last_state_change_key(id)) + shared_buffer:delete(gen_half_open_calls_key(id)) + shared_buffer:delete(gen_half_open_success_key(id)) + shared_buffer:delete(gen_half_open_completed_key(id)) + -- give the route a fresh start instead of reopening on pre-recovery history + clear_window(conf, id) +end + +local function break_response(conf, ctx) + if conf.break_response_body then + if conf.break_response_headers then + for _, value in ipairs(conf.break_response_headers) do + local val = core.utils.resolve_var(value.value, ctx.var) + core.response.add_header(value.key, val) + end + end + return conf.break_response_code, conf.break_response_body + end + return conf.break_response_code +end + +-- Count-based circuit breaker (based on latest APISIX version) +local function count_based_access(conf, ctx) local unhealthy_key = gen_unhealthy_key(ctx) -- unhealthy counts local unhealthy_count, err = shared_buffer:get(unhealthy_key) @@ -180,23 +464,86 @@ function _M.access(conf, ctx) -- breaker if lasttime + breaker_time >= ngx.time() then - if conf.break_response_body then - if conf.break_response_headers then - for _, value in ipairs(conf.break_response_headers) do - local val = core.utils.resolve_var(value.value, ctx.var) - core.response.add_header(value.key, val) - end - end - return conf.break_response_code, conf.break_response_body + return break_response(conf, ctx) + end + + return +end + +-- Ratio-based circuit breaker +local function ratio_based_access(conf, ctx) + local id = gen_breaker_id(ctx) + local current_state = get_circuit_breaker_state(id) + + if current_state == OPEN then + local last_change_time, err = shared_buffer:get(gen_last_state_change_key(id)) + if err then + core.log.warn("failed to get last change time: ", err) + return break_response(conf, ctx) + end + + local wait_duration = conf.max_breaker_sec or 300 + if not last_change_time or (ngx.time() - last_change_time) < wait_duration then + return break_response(conf, ctx) + end + + -- Try to become the single request that flips OPEN -> HALF_OPEN. The + -- lock's own short TTL is the debounce window: any other request + -- that also observed OPEN and raced in here gets conservatively + -- rejected as still-open rather than risking an admitted-but- + -- uncounted probe slipping past half_open_max_calls. + local won, add_err = shared_buffer:add(gen_transition_lock_key(id), 1, 1) + if add_err then + core.log.warn("failed to add transition lock: ", add_err) + end + + if not won then + return break_response(conf, ctx) + end + + set_circuit_breaker_state(conf, id, HALF_OPEN) + local half_open_ttl = get_state_ttl(conf) + shared_buffer:set(gen_half_open_calls_key(id), 0, half_open_ttl) + shared_buffer:set(gen_half_open_success_key(id), 0, half_open_ttl) + shared_buffer:set(gen_half_open_completed_key(id), 0, half_open_ttl) + core.log.info("api-breaker: transitioned from OPEN to HALF_OPEN for ", id) + + -- Fall through so this request is admitted (and counted) as the + -- first half-open probe, instead of getting a free, uncounted pass. + current_state = HALF_OPEN + end + + if current_state == HALF_OPEN then + local permitted_calls = conf.unhealthy.half_open_max_calls or 3 + local half_open_calls, err = shared_buffer:incr( + gen_half_open_calls_key(id), 1, 0, get_state_ttl(conf)) + if err then + core.log.warn("failed to increment half-open calls: ", err) end - return conf.break_response_code + + if half_open_calls > permitted_calls then + return break_response(conf, ctx) + end + + return end + -- CLOSED: allow the request through; the decision to trip the breaker + -- is made in the log phase once the response outcome is known. return end +function _M.access(conf, ctx) + if conf.policy == "unhealthy-ratio" then + return ratio_based_access(conf, ctx) + else + -- Default to count-based (unhealthy-count) + return count_based_access(conf, ctx) + end +end -function _M.log(conf, ctx) +-- Count-based logging (based on latest APISIX version) +local function count_based_log(conf, ctx) local unhealthy_key = gen_unhealthy_key(ctx) local healthy_key = gen_healthy_key(ctx) local upstream_status = core.response.get_upstream_status(ctx) @@ -264,4 +611,99 @@ function _M.log(conf, ctx) return end +-- Ratio-based logging +local function ratio_based_log(conf, ctx) + local upstream_status = core.response.get_upstream_status(ctx) + if not upstream_status then + return + end + + local id = gen_breaker_id(ctx) + local current_state = get_circuit_breaker_state(id) + + local is_failure = core.table.array_find(conf.unhealthy.http_statuses, upstream_status) + local is_success = not is_failure and + core.table.array_find(conf.healthy.http_statuses, upstream_status) + + if current_state == HALF_OPEN then + if is_failure then + -- Fail fast: any failed probe reopens the breaker immediately. + core.log.warn("api-breaker: half-open probe failed for ", id, + ", reopening") + open_breaker(conf, id) + return + end + + if is_success then + shared_buffer:incr(gen_half_open_success_key(id), 1, 0, get_state_ttl(conf)) + end + + -- Count every probe that reaches the log phase, regardless of how + -- its status was classified, so a status that's neither a + -- configured success nor failure can't strand the breaker in + -- HALF_OPEN forever waiting for a classification that will never + -- come. Probes that the access phase rejected for exceeding + -- half_open_max_calls never reach here, so they can't corrupt this + -- count either. + local permitted_calls = conf.unhealthy.half_open_max_calls or 3 + local completed, err = shared_buffer:incr( + gen_half_open_completed_key(id), 1, 0, get_state_ttl(conf)) + if err then + core.log.warn("failed to increment half-open completed calls: ", err) + return + end + + if completed >= permitted_calls then + local success_count = shared_buffer:get(gen_half_open_success_key(id)) or 0 + local success_ratio = (conf.healthy and conf.healthy.success_ratio) or 0.6 + + if (success_count / completed) >= success_ratio then + core.log.info("api-breaker: half-open success ratio ", + success_count / completed, " >= threshold ", success_ratio, + ", closing for ", id) + close_breaker(conf, id) + else + core.log.warn("api-breaker: half-open success ratio ", + success_count / completed, " < threshold ", success_ratio, + ", reopening for ", id) + open_breaker(conf, id) + end + end + + return + end + + if current_state == OPEN then + -- access() should already be blocking requests while OPEN; avoid + -- polluting the closed-state window if one still slips through. + return + end + + -- CLOSED: record this response and check whether the error ratio over + -- the sliding window now crosses the configured threshold. + record_window_sample(conf, id, is_failure) + + local min_requests = conf.unhealthy.min_request_threshold or 10 + local error_ratio = conf.unhealthy.error_ratio or 0.5 + local total_requests, total_failures = sum_window(conf, id) + + if total_requests >= min_requests then + local failure_rate = total_failures / total_requests + if failure_rate >= error_ratio then + core.log.warn("api-breaker: error ratio ", failure_rate, + " >= threshold ", error_ratio, ", opening for ", id) + open_breaker(conf, id) + end + end +end + +function _M.log(conf, ctx) + if conf.policy == "unhealthy-ratio" then + ratio_based_log(conf, ctx) + else + -- Default to count-based (unhealthy-count) + count_based_log(conf, ctx) + end +end + return _M diff --git a/docs/en/latest/plugins/api-breaker.md b/docs/en/latest/plugins/api-breaker.md index 3f2384323aff..a9ef94a55b7e 100644 --- a/docs/en/latest/plugins/api-breaker.md +++ b/docs/en/latest/plugins/api-breaker.md @@ -30,14 +30,31 @@ description: This document describes the information about the Apache APISIX api The `api-breaker` Plugin implements circuit breaker functionality to protect Upstream services. +This plugin supports two circuit breaker policies: + +- **Failure count-based circuit breaking (`unhealthy-count`)**: Triggers circuit breaker when consecutive failure count reaches the threshold +- **Error ratio-based circuit breaking (`unhealthy-ratio`)**: Triggers circuit breaker when error rate within a sliding time window reaches the threshold + :::note +**Failure count-based circuit breaking (`unhealthy-count`)**: + Whenever the Upstream service responds with a status code from the configured `unhealthy.http_statuses` list for the configured `unhealthy.failures` number of times, the Upstream service will be considered unhealthy. The request is then retried in 2, 4, 8, 16 ... seconds until the `max_breaker_sec`. In an unhealthy state, if the Upstream service responds with a status code from the configured list `healthy.http_statuses` for `healthy.successes` times, the service is considered healthy again. +**Error ratio-based circuit breaking (`unhealthy-ratio`)**: + +This policy is based on sliding time window statistics for error rate. When the total number of requests reaches `min_request_threshold` and the error rate exceeds `error_ratio` within the `sliding_window_size` time window, the circuit breaker enters the open state for `max_breaker_sec` seconds. + +The circuit breaker has three states: + +- **CLOSED**: Normal request forwarding +- **OPEN**: Directly returns circuit breaker response without forwarding requests +- **HALF_OPEN**: Allows a limited number of requests to test if the service has recovered + ::: ## Attributes @@ -47,15 +64,35 @@ In an unhealthy state, if the Upstream service responds with a status code from | break_response_code | integer | True | | [200, ..., 599] | HTTP error code to return when Upstream is unhealthy. | | break_response_body | string | False | | | Body of the response message to return when Upstream is unhealthy. | | break_response_headers | array[object] | False | | [{"key":"header_name","value":"can contain Nginx $var"}] | Headers of the response message to return when Upstream is unhealthy. Can only be configured when the `break_response_body` attribute is configured. The values can contain APISIX variables. For example, we can use `{"key":"X-Client-Addr","value":"$remote_addr:$remote_port"}`. | -| max_breaker_sec | integer | False | 300 | >=3 | Maximum time in seconds for circuit breaking. | +| max_breaker_sec | integer | False | 300 | >=3 | Maximum time in seconds for circuit breaking. Applies to both circuit breaker policies. | +| policy | string | False | "unhealthy-count" | ["unhealthy-count", "unhealthy-ratio"] | Circuit breaker policy. `unhealthy-count` for failure count-based circuit breaking, `unhealthy-ratio` for error ratio-based circuit breaking. | + +### Failure count-based circuit breaking (policy = "unhealthy-count") + +| Name | Type | Required | Default | Valid values | Description | +|-------------------------|----------------|----------|---------|-----------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | unhealthy.http_statuses | array[integer] | False | [500] | [500, ..., 599] | Status codes of Upstream to be considered unhealthy. | | unhealthy.failures | integer | False | 3 | >=1 | Number of failures within a certain period of time for the Upstream service to be considered unhealthy. | | healthy.http_statuses | array[integer] | False | [200] | [200, ..., 499] | Status codes of Upstream to be considered healthy. | | healthy.successes | integer | False | 3 | >=1 | Number of consecutive healthy requests for the Upstream service to be considered healthy. | +### Error ratio-based circuit breaking (policy = "unhealthy-ratio") + +| Name | Type | Required | Default | Valid values | Description | +|-------------------------|----------------|----------|---------|-----------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| +| unhealthy.http_statuses | array[integer] | False | [500] | [500, ..., 599] | Status codes of Upstream to be considered unhealthy. | +| unhealthy.error_ratio | number | False | 0.5 | [0, 1] | Error rate threshold to trigger circuit breaker. For example, 0.5 means circuit breaker triggers when error rate reaches 50%. | +| unhealthy.min_request_threshold | integer | False | 10 | >=1 | Minimum number of requests required within the sliding window to trigger circuit breaker. Circuit breaker will only evaluate error rate when request count reaches this threshold. | +| unhealthy.sliding_window_size | integer | False | 300 | [10, 3600] | Size of the sliding window in seconds. The time range used to calculate error rate. | +| unhealthy.half_open_max_calls | integer | False | 3 | [1, 20] | Number of permitted calls when circuit breaker is in half-open state. Used to test if the service has recovered. | +| healthy.http_statuses | array[integer] | False | [200] | [200, ..., 499] | Status codes of Upstream to be considered healthy. | +| healthy.success_ratio | number | False | 0.6 | [0, 1] | Success rate threshold to close circuit breaker from half-open state. For example, 0.6 means circuit breaker closes when success rate reaches 60%. | + ## Enable Plugin -The example below shows how you can configure the Plugin on a specific Route: +### Failure count-based circuit breaking example + +The example below shows how you can configure the Plugin with failure count-based circuit breaking policy on a specific Route: :::note You can fetch the `admin_key` from `config.yaml` and save to an environment variable with the following command: @@ -73,6 +110,7 @@ curl "http://127.0.0.1:9180/apisix/admin/routes/1" \ "plugins": { "api-breaker": { "break_response_code": 502, + "policy": "unhealthy-count", "unhealthy": { "http_statuses": [500, 503], "failures": 3 @@ -95,6 +133,47 @@ curl "http://127.0.0.1:9180/apisix/admin/routes/1" \ In this configuration, a response code of `500` or `503` three times within a certain period of time triggers the unhealthy status of the Upstream service. A response code of `200` restores its healthy status. +### Error ratio-based circuit breaking example + +The example below shows how to enable error ratio-based circuit breaking policy. This configuration triggers circuit breaker when the request count reaches 10 and error rate exceeds 50% within a 5-minute sliding window: + +```shell +curl "http://127.0.0.1:9180/apisix/admin/routes/2" \ +-H "X-API-KEY: $admin_key" -X PUT -d ' +{ + "plugins": { + "api-breaker": { + "break_response_code": 503, + "break_response_body": "Service temporarily unavailable due to high error rate", + "break_response_headers": [ + {"key": "X-Circuit-Breaker", "value": "open"}, + {"key": "Retry-After", "value": "60"} + ], + "policy": "unhealthy-ratio", + "max_breaker_sec": 60, + "unhealthy": { + "http_statuses": [500, 502, 503, 504], + "error_ratio": 0.5, + "min_request_threshold": 10, + "sliding_window_size": 300, + "half_open_max_calls": 3 + }, + "healthy": { + "http_statuses": [200, 201, 202], + "success_ratio": 0.6 + } + } + }, + "upstream": { + "type": "roundrobin", + "nodes": { + "127.0.0.1:1980": 1 + } + }, + "uri": "/api" +}' +``` + ## Example usage Once you have configured the Plugin as shown above, you can test it out by sending a request. diff --git a/docs/zh/latest/plugins/api-breaker.md b/docs/zh/latest/plugins/api-breaker.md index 5183f84763e7..3180b2713494 100644 --- a/docs/zh/latest/plugins/api-breaker.md +++ b/docs/zh/latest/plugins/api-breaker.md @@ -6,7 +6,6 @@ keywords: - API Breaker description: 本文介绍了 Apache APISIX api-breaker 插件的相关操作，你可以使用此插件的 API 熔断机制来保护上游业务服务。 --- -