diff --git a/apisix/balancer.lua b/apisix/balancer.lua index 35f015da1b45..0258d4bb54d0 100644 --- a/apisix/balancer.lua +++ b/apisix/balancer.lua @@ -20,6 +20,7 @@ local core = require("apisix.core") local priority_balancer = require("apisix.balancer.priority") local apisix_upstream = require("apisix.upstream") local healthcheck_manager = require("apisix.healthcheck_manager") +local slow_start = require("apisix.slow_start") local ipairs = ipairs local is_http = ngx.config.subsystem == "http" local enable_keepalive = balancer.enable_keepalive and is_http @@ -50,7 +51,7 @@ local _M = { } -local function transform_node(new_nodes, node) +local function transform_node(new_nodes, node, weight) if not new_nodes._priority_index then new_nodes._priority_index = {} end @@ -60,16 +61,17 @@ local function transform_node(new_nodes, node) core.table.insert(new_nodes._priority_index, node.priority) end - new_nodes[node.priority][node.host .. ":" .. node.port] = node.weight + new_nodes[node.priority][node.host .. ":" .. node.port] = weight or node.weight return new_nodes end -local function fetch_all_nodes(upstream) - local nodes = upstream.nodes +-- `weights` carries the slow start weight of every node, indexed like `nodes`; +-- without it each node keeps its configured weight +local function transform_nodes(nodes, weights) local new_nodes = core.table.new(0, #nodes) - for _, node in ipairs(nodes) do - new_nodes = transform_node(new_nodes, node) + for i, node in ipairs(nodes) do + new_nodes = transform_node(new_nodes, node, weights and weights[i]) end return new_nodes end @@ -107,26 +109,27 @@ local function create_health_status(upstream, checker) end --- Build the picker node set from the healthy subset, reusing create_health_status --- so the per-node health lookup lives in exactly one place. -local function fetch_health_nodes(upstream, checker) +-- The nodes that actually reach the picker, reusing create_health_status so the +-- per-node health lookup lives in exactly one place. When every node is unhealthy +-- the whole set is kept, which is the existing fail-open behaviour. +local function fetch_eligible_nodes(upstream, checker) if not checker then - return fetch_all_nodes(upstream) + return upstream.nodes end local health_status = create_health_status(upstream, checker) if health_status.all_unhealthy then - return fetch_all_nodes(upstream) + return upstream.nodes end - local up_nodes = core.table.new(0, #upstream.nodes) + local nodes = core.table.new(#upstream.nodes, 0) for _, node in ipairs(upstream.nodes) do if health_status.status[node.host .. ":" .. node.port] then - up_nodes = transform_node(up_nodes, node) + core.table.insert(nodes, node) end end - return up_nodes + return nodes end @@ -164,9 +167,12 @@ local function create_server_picker(upstream, checker) local up_nodes if upstream.type == "chash" then - up_nodes = fetch_all_nodes(upstream) + up_nodes = transform_nodes(upstream.nodes) else - up_nodes = fetch_health_nodes(upstream, checker) + -- slow start runs on the eligible set, so a node only starts its ramp + -- once it can actually be picked + local nodes = fetch_eligible_nodes(upstream, checker) + up_nodes = transform_nodes(nodes, slow_start.effective_weights(upstream, nodes)) end if #up_nodes._priority_index > 1 then @@ -256,7 +262,11 @@ local function pick_server(route, ctx) -- balancer here would leave it blind to everything routed before the second -- node showed up, which is the state a k8s deployment or a discovery service -- starts from. See #12217 - if nodes_count == 1 and up_conf.type ~= "least_conn" then + -- + -- Slow start is in the same position: the node set of a single node upstream + -- is what the second node is later compared against, and only the picker build + -- records it. The node still takes every request either way. + if nodes_count == 1 and up_conf.type ~= "least_conn" and not up_conf.warm_up_conf then local node = up_conf.nodes[1] ctx.balancer_ip = node.host ctx.balancer_port = node.port @@ -307,6 +317,10 @@ local function pick_server(route, ctx) version = version .. "#" .. checker.status_ver end + if up_conf.warm_up_conf then + version = version .. (slow_start.version_suffix(up_conf) or "") + end + -- the same picker will be used in the whole request, especially during the retry local server_picker = ctx.server_picker if not server_picker then diff --git a/apisix/cli/config.lua b/apisix/cli/config.lua index f4c002835396..29d4b430b808 100644 --- a/apisix/cli/config.lua +++ b/apisix/cli/config.lua @@ -168,6 +168,7 @@ local _M = { ["plugin-limit-conn"] = "10m", ["worker-events"] = "10m", ["lrucache-lock"] = "10m", + ["upstream-slow-start"] = "10m", ["balancer-ewma"] = "10m", ["balancer-ewma-locks"] = "10m", ["balancer-ewma-last-touched-at"] = "10m", diff --git a/apisix/cli/ngx_tpl.lua b/apisix/cli/ngx_tpl.lua index 42833de45835..80dc1312d073 100644 --- a/apisix/cli/ngx_tpl.lua +++ b/apisix/cli/ngx_tpl.lua @@ -408,6 +408,7 @@ http { lua_shared_dict internal-status {* http.lua_shared_dict["internal-status"] *}; lua_shared_dict worker-events {* http.lua_shared_dict["worker-events"] *}; lua_shared_dict lrucache-lock {* http.lua_shared_dict["lrucache-lock"] *}; + lua_shared_dict upstream-slow-start {* http.lua_shared_dict["upstream-slow-start"] *}; lua_shared_dict balancer-ewma {* http.lua_shared_dict["balancer-ewma"] *}; lua_shared_dict balancer-ewma-locks {* http.lua_shared_dict["balancer-ewma-locks"] *}; lua_shared_dict balancer-ewma-last-touched-at {* http.lua_shared_dict["balancer-ewma-last-touched-at"] *}; diff --git a/apisix/init.lua b/apisix/init.lua index 20d0caa1f1d5..b4ef63195e84 100644 --- a/apisix/init.lua +++ b/apisix/init.lua @@ -94,6 +94,7 @@ function _M.http_init(args) core.resolver.init_resolver(args) core.id.init() core.env.init() + require("apisix.slow_start").init() local process = require("ngx.process") local ok, err = process.enable_privileged_agent() diff --git a/apisix/plugins/traffic-split.lua b/apisix/plugins/traffic-split.lua index da6014e8b5f3..35243f502c94 100644 --- a/apisix/plugins/traffic-split.lua +++ b/apisix/plugins/traffic-split.lua @@ -117,6 +117,16 @@ function _M.check_schema(conf) end end end + + -- the upstreams this plugin builds are rebuilt per request and carry + -- no stable scope, so slow start has nowhere to keep the lifecycle of + -- their nodes + for _, wupstream in ipairs(rule.weighted_upstreams or {}) do + if wupstream.upstream and wupstream.upstream.warm_up_conf then + return false, "warm_up_conf is not supported by the upstream of " .. + "the traffic-split plugin" + end + end end end diff --git a/apisix/schema_def.lua b/apisix/schema_def.lua index 85dd40905299..c36e7cdddd88 100644 --- a/apisix/schema_def.lua +++ b/apisix/schema_def.lua @@ -414,6 +414,48 @@ local private_key_schema = { } +local warm_up_conf_schema = { + description = "slow start: ramp a newly observed node up to its configured weight", + type = "object", + properties = { + slow_start_time_seconds = { + description = "seconds a new node takes to reach its full weight", + type = "integer", + minimum = 1, + }, + min_weight_percent = { + description = "lowest effective weight, as a percentage of the original weight", + type = "integer", + minimum = 1, + maximum = 100, + }, + interval = { + description = "seconds between two effective weight refreshes", + type = "integer", + minimum = 1, + default = 1, + }, + aggression = { + description = "shape of the ramp: 1 is linear, above 1 ramps up faster " .. + "at the beginning, below 1 slower", + type = "number", + minimum = 0.01, + default = 1, + }, + startup_grace_period_seconds = { + description = "seconds after the data plane started during which a node " .. + "observed for the first time is considered mature", + type = "integer", + minimum = 0, + default = 0, + }, + }, + required = {"slow_start_time_seconds", "min_weight_percent"}, + additionalProperties = false, +} +_M.warm_up_conf = warm_up_conf_schema + + local upstream_schema = { type = "object", properties = { @@ -427,6 +469,7 @@ local upstream_schema = { -- properties nodes = nodes_schema, + warm_up_conf = warm_up_conf_schema, retries = { type = "integer", minimum = 0, diff --git a/apisix/slow_start.lua b/apisix/slow_start.lua new file mode 100644 index 000000000000..1ca6e8f1f32d --- /dev/null +++ b/apisix/slow_start.lua @@ -0,0 +1,531 @@ +-- +-- Licensed to the Apache Software Foundation (ASF) under one or more +-- contributor license agreements. See the NOTICE file distributed with +-- this work for additional information regarding copyright ownership. +-- The ASF licenses this file to You under the Apache License, Version 2.0 +-- (the "License"); you may not use this file except in compliance with +-- the License. You may obtain a copy of the License at +-- +-- http://www.apache.org/licenses/LICENSE-2.0 +-- +-- Unless required by applicable law or agreed to in writing, software +-- distributed under the License is distributed on an "AS IS" BASIS, +-- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +-- See the License for the specific language governing permissions and +-- limitations under the License. +-- +-- Slow start (`upstream.warm_up_conf`): a node that the data plane observes for +-- the first time takes a reduced share of the traffic and ramps back to its +-- configured weight over `slow_start_time_seconds`. +-- +-- Which node is new is decided here, by comparing the node set of the current +-- picker build against the set the previous build recorded. No node lifecycle +-- timestamp is read from the configuration, the Admin API or service discovery: +-- every start point is generated locally with `ngx.now()` and lives only in the +-- `upstream-slow-start` shared dict, so it is never written back to etcd. +-- +-- Shared dict layout, all keys prefixed by the upstream scope (the resource key +-- of the standalone upstream, or of the route/service that embeds it): +-- +-- |# node ids the configuration held at the previous reconcile +-- |! aggregate ramp deadline, read on the request hot path +-- | ramp start: > 0 ramping, 0 mature, -1 not picked yet +-- ||i when a reconcile first saw it configured but unpickable +-- |@ the successor a worker elected for that ramp start +-- +-- Workers reconcile in parallel, and the shared dict has no compare-and-set, so +-- a value is never derived from a read and then written back blindly: a worker +-- that read a ramp start just before another one changed it would put the old +-- one back. Instead a node's ramp start is only ever created with `add`, only +-- ever replaced through an election on `add` keyed by the value being replaced, +-- and otherwise kept alive with `expire`, which leaves the value alone. Every +-- worker that decides on a change to the same ramp start therefore ends up +-- writing the same successor. +-- +-- Presence and eligibility are tracked separately. Presence comes from the +-- configuration, so every worker sees the same set; eligibility is this worker's +-- health view and only decides when a ramp starts and when it is interrupted. +-- Tombstoning off the eligible set would let one worker's transient health +-- opinion drop a node the other workers are still serving, and the state of a +-- node that never left could then expire and be ramped again from scratch. +-- +-- A node the configuration still holds keeps its entry alive. One that leaves the +-- configuration keeps it for a further `slow_start_time_seconds` (the tombstone +-- window) so a short absence resumes the ramp instead of restarting it. Coming +-- back later, or being out of the picker for longer than a full window, starts a +-- new lifecycle. +local core = require("apisix.core") +local ipairs = ipairs +local pairs = pairs +local tostring = tostring +local tonumber = tonumber +local os_time = os.time +local ngx_now = ngx.now +local math_floor = math.floor +local math_max = math.max +local math_min = math.min +local str_gmatch = string.gmatch +local type = type + +local shdict = ngx.shared["upstream-slow-start"] + +local INSTANCE_STARTED_KEY = "@instance_started_at" +local SNAPSHOT_SUFFIX = "|#" +local DEADLINE_SUFFIX = "|!" +local INELIGIBLE_SUFFIX = "|i" + +-- A node that is present but needs no ramp is stored with this ramp start, so +-- that it can still be tombstoned when it disappears: without an entry it would +-- look new again on its way back. +local MATURE = 0 + +-- And one the configuration holds but that has never reached a picker - an +-- active health check has not cleared it yet - is stored with this one. It is +-- what separates "known and already mature" from "known but has not had its +-- chance to ramp", which otherwise both look like a node with no state. +local PENDING = -1 + +-- Lifetime of the state of a node that is still eligible. Refreshed on every +-- reconcile, which a busy upstream runs at least as often as the picker LRU +-- expires (300s), so it only ever elapses for an upstream that stopped receiving +-- traffic altogether. State lost that way makes the whole node set a fresh mature +-- baseline again, which is the safe direction, and it is also what keeps the +-- state of a deleted upstream from living forever. +local STATE_TTL = 86400 + +-- Lifetime of an election on a ramp start, see `replace_start`. +local ELECTION_TTL = 10 + +local _M = {} + + +local reported_scopes = {} + + +local function report_once(scope, ...) + if reported_scopes[scope] then + return + end + reported_scopes[scope] = true + core.log.error(...) +end + + +-- The instance start point behind `startup_grace_period_seconds`. Written from +-- the master process with `add`, so a HUP reload - which keeps the shared dict - +-- continues to use the start point of the original start. +function _M.init() + if not shdict then + return + end + + local ok, err = shdict:add(INSTANCE_STARTED_KEY, os_time()) + if not ok and err ~= "exists" then + core.log.error("failed to record the slow start instance start time: ", err) + end +end + + +local function scope_key(up_conf) + return up_conf.resource_key +end + + +-- The lifecycle identity of a node. A domain node is identified by the hostname +-- it was configured with, not by the address it currently resolves to, so a DNS +-- rotation does not restart its ramp. +local function node_id(node) + return (node.domain or node.host) .. ":" .. tostring(node.port) +end + + +local function effective_weight(original_weight, first_seen_at, conf, now) + if original_weight <= 0 then + -- a node configured out of the rotation stays out of it + return original_weight + end + + if not first_seen_at or first_seen_at <= MATURE then + return original_weight + end + + local window = conf.slow_start_time_seconds + local elapsed = now - first_seen_at + if elapsed < 0 then + -- the local clock moved backwards; start the window over rather than + -- letting a negative elapsed produce a complex power below + elapsed = 0 + end + if elapsed >= window then + return original_weight + end + + -- `max(elapsed, 1)` keeps the first bucket at one second of progress, so a + -- node never ramps from exactly zero + local time_factor = math_max(elapsed, 1) / window + local ratio = math_max(conf.min_weight_percent / 100, + time_factor ^ (1 / (conf.aggression or 1))) + local weight = math_floor(original_weight * ratio) + if weight < 1 then + -- integer weights: anything above zero has to keep at least one unit, + -- otherwise the node gets no traffic at all and never warms up + weight = 1 + end + + return weight +end +_M.effective_weight = effective_weight + + +local function within_startup_grace(conf, now) + local grace = conf.startup_grace_period_seconds or 0 + if grace <= 0 then + return false + end + + local started_at = shdict:get(INSTANCE_STARTED_KEY) + if not started_at then + return false + end + + return now < started_at + grace +end + + +local function decode_snapshot(snapshot) + local known = {} + if not snapshot then + return nil + end + + for id in str_gmatch(snapshot, "[^,]+") do + known[id] = true + end + return known +end + + +local function store(key, value, ttl) + local ok, err, forcible = shdict:set(key, value, ttl) + if not ok then + core.log.error("failed to store the slow start state of ", key, ": ", err) + elseif forcible then + core.log.warn("the upstream-slow-start shared dict is full, storing ", key, + " evicted another entry; nodes whose state is lost keep ", + "their configured weight") + end + return ok +end + + +local function keep_alive(key, ttl) + local ok, err = shdict:expire(key, ttl) + if not ok and err ~= "not found" then + core.log.error("failed to refresh the slow start state of ", key, ": ", err) + end +end + + +-- Create the ramp start of a node that has none, or adopt the one another worker +-- created first. Returns the ramp start everyone now shares, and whether this +-- worker is the one that set it. +local function create_start(key, candidate) + local ok, err, forcible = shdict:add(key, candidate, STATE_TTL) + if ok then + if forcible then + core.log.warn("the upstream-slow-start shared dict is full, storing ", key, + " evicted another entry") + end + return candidate, true + end + + if err ~= "exists" then + core.log.error("failed to store the slow start state of ", key, ": ", err) + return MATURE, false + end + + return tonumber(shdict:get(key)) or candidate, false +end + + +-- Replace the ramp start `from` with `to`, unless another worker already elected +-- a successor for `from`, in which case that one is used. The election outlives +-- the few microseconds a worker spends between reading `from` and getting here +-- by a wide margin, and is short enough that `from` cannot come round again for +-- the same node before it expires. +local function replace_start(key, from, to, window) + local election = key .. "@" .. from + local won, err = shdict:add(election, to, math_min(window, ELECTION_TTL)) + if not won then + if err == "exists" then + to = tonumber(shdict:get(election)) or to + else + core.log.error("failed to elect the slow start state of ", key, ": ", err) + end + end + + store(key, to, STATE_TTL) + return to, won +end + + +-- Compare the node set of this picker build against the one the previous build +-- recorded, and return the ramp start point of every eligible node, indexed like +-- `nodes`. `present_nodes` is the configured set, identical in every worker; +-- `nodes` is the subset this worker can actually pick from. +local function reconcile(conf, present_nodes, nodes, scope, now) + local window = conf.slow_start_time_seconds + + local present_ids = core.table.new(#present_nodes, 0) + local present = core.table.new(0, #present_nodes) + for i, node in ipairs(present_nodes) do + present_ids[i] = node_id(node) + present[present_ids[i]] = true + end + + local ids = core.table.new(#nodes, 0) + local eligible = core.table.new(0, #nodes) + for i, node in ipairs(nodes) do + ids[i] = node_id(node) + eligible[ids[i]] = true + end + + local snapshot_key = scope .. SNAPSHOT_SUFFIX + local known = decode_snapshot(shdict:get(snapshot_key)) + + -- The first reconcile of a scope is a baseline: the nodes an upstream is + -- bootstrapped with, and the nodes it already had when `warm_up_conf` was + -- turned on, are mature. So are nodes observed inside the startup grace + -- period, which absorbs the ordering differences of a cold restart. + local baseline = (known == nil) or within_startup_grace(conf, now) + + local first_seen = core.table.new(#nodes, 0) + local deadline = 0 + + local function log_began(id, weight) + core.log.info("slow start began for node ", id, " of upstream ", scope, + ", weight ", weight, ", window ", window, "s, from ", now) + end + + for i, id in ipairs(ids) do + local key = scope .. "|" .. id + local ineligible_key = key .. INELIGIBLE_SUFFIX + local start_at = tonumber(shdict:get(key)) + local ineligible_since = tonumber(shdict:get(ineligible_key)) + local changed + + if not start_at then + if baseline or (known and known[id]) then + -- either the bootstrap set, or a node whose state the shared dict + -- evicted while it stayed configured: both keep the full weight + -- rather than ramping a node that has been serving all along + start_at = create_start(key, MATURE) + else + start_at, changed = create_start(key, now) + if changed then + log_began(id, nodes[i].weight) + end + end + + elseif start_at == PENDING then + -- the first picker it can actually be part of: this is where its + -- window starts, not when the configuration first mentioned it + start_at, changed = replace_start(key, PENDING, now, window) + if changed then + log_began(id, nodes[i].weight) + end + + elseif ineligible_since and now - ineligible_since > window then + -- observed out of the picker for longer than a full window: whatever + -- answers on this address now is not the process that was ramping + local out_for = now - ineligible_since + start_at, changed = replace_start(key, start_at, now, window) + if changed then + core.log.info("slow start restarted for node ", id, " of upstream ", + scope, " after ", out_for, "s out of the picker") + end + + elseif start_at > MATURE and now - start_at >= window then + start_at, changed = replace_start(key, start_at, MATURE, window) + if changed then + core.log.info("slow start finished for node ", id, " of upstream ", + scope) + end + + else + -- refresh the lifetime, which also drops the tombstone window if the + -- node came back after having left the configuration + keep_alive(key, STATE_TTL) + end + + if ineligible_since then + -- back in the picker + shdict:delete(ineligible_key) + end + + first_seen[i] = start_at + if start_at > MATURE then + deadline = math_max(deadline, start_at + window) + end + end + + for _, id in ipairs(present_ids) do + if not eligible[id] then + local key = scope .. "|" .. id + local start_at = tonumber(shdict:get(key)) + if start_at then + -- still configured, just not pickable here: keep the state alive + -- and mark when it dropped out, keeping the earliest mark. The + -- ramp clock itself runs on, so a short outage resumes where it + -- left off + keep_alive(key, STATE_TTL) + local ineligible_key = key .. INELIGIBLE_SUFFIX + if not shdict:add(ineligible_key, now, STATE_TTL) then + keep_alive(ineligible_key, STATE_TTL) + end + elseif baseline or (known and known[id]) then + -- part of the bootstrap set even though a health check has not + -- cleared it yet, or state the shared dict evicted: every node + -- the previous reconcile saw was given state then + start_at = create_start(key, MATURE) + else + -- configured but never picked: its window starts when it first + -- becomes usable + start_at = create_start(key, PENDING) + end + + -- a ramp this worker cannot pick from is still a ramp other workers + -- may be serving, and the deadline is what keeps their picker keys + -- moving; leaving it out would let this worker publish an early one + if start_at > MATURE then + deadline = math_max(deadline, start_at + window) + end + end + end + + -- A node that left the configuration keeps its state for one slow start + -- window and is then forgotten, so that the address coming back later - a + -- different process behind the same host and port - is warmed up again. + if known then + for id in pairs(known) do + if not present[id] then + local key = scope .. "|" .. id + if shdict:get(key) then + core.log.info("node ", id, " of upstream ", scope, + " left the upstream, keeping its slow start state for ", + window, "s") + keep_alive(key, window) + keep_alive(key .. INELIGIBLE_SUFFIX, window) + end + end + end + end + + core.table.sort(present_ids) + store(snapshot_key, core.table.concat(present_ids, ","), STATE_TTL) + -- published for the hot path: zero means every node is mature + store(scope .. DEADLINE_SUFFIX, deadline, STATE_TTL) + + return first_seen +end + + +local function usable(up_conf, nodes) + local scope = scope_key(up_conf) + if not scope then + core.log.error("slow start needs an upstream with a resource key, ", + "ignoring warm_up_conf") + return nil + end + + if not shdict then + -- slow start only ramps HTTP upstreams, and the stream subsystem has no + -- such shared dict. Like every other upstream field that does not apply + -- there, `warm_up_conf` is quietly ignored and the configured weights are + -- used, rather than failing the connection or logging on every build + return nil + end + + if up_conf.type ~= "roundrobin" then + report_once(scope, "slow start only supports roundrobin, ignoring ", + "warm_up_conf of upstream ", scope) + return nil + end + + if nodes then + local priority = nodes[1] and nodes[1].priority + for _, node in ipairs(nodes) do + if node.priority ~= priority then + report_once(scope, "slow start does not support an upstream with ", + "mixed node priorities, ignoring warm_up_conf of upstream ", + scope) + return nil + end + end + end + + return scope +end + + +-- Effective weight of every node of this picker build, indexed like `nodes`, or +-- nil when the upstream does not use slow start. Runs once per picker build, not +-- per request. +function _M.effective_weights(up_conf, nodes) + local conf = up_conf.warm_up_conf + if type(conf) ~= "table" then + return nil + end + + local scope = usable(up_conf, nodes) + if not scope then + -- settle the picker cache key: without a deadline `version_suffix` keeps + -- appending a fresh time bucket, so an upstream that can never ramp would + -- rebuild its picker every `interval` for nothing + if shdict and up_conf.resource_key then + store(up_conf.resource_key .. DEADLINE_SUFFIX, 0, STATE_TTL) + end + return nil + end + + local now = ngx_now() + local first_seen = reconcile(conf, up_conf.nodes, nodes, scope, now) + + local weights = core.table.new(#nodes, 0) + for i, node in ipairs(nodes) do + weights[i] = effective_weight(node.weight, first_seen[i], conf, now) + end + + return weights +end + + +-- Suffix of the picker cache key. While any node ramps, the key carries the +-- current `interval` bucket so that the picker is rebuilt once per bucket per +-- worker; once every node is mature the key settles on a stable suffix, and the +-- rebuild it causes is the one that restores the full weights. +-- +-- A rebuild restarts the round robin cursor, which favours the heaviest node for +-- the first few picks. Over a bucket that carries real traffic this averages out, +-- but on an upstream that sees only a handful of requests per `interval` the +-- ramping node can end up with even less traffic than its weight asks for. +function _M.version_suffix(up_conf) + local conf = up_conf.warm_up_conf + if type(conf) ~= "table" or not shdict or not up_conf.resource_key + or up_conf.type ~= "roundrobin" then + return nil + end + + local now = ngx_now() + local deadline = shdict:get(up_conf.resource_key .. DEADLINE_SUFFIX) + -- an unknown deadline means no picker has been built for this upstream yet; + -- the build that this bucket triggers is the one that publishes it + if deadline and now >= deadline then + return "#wm" + end + + return "#w" .. math_floor(now / (conf.interval or 1)) +end + + +return _M diff --git a/apisix/upstream.lua b/apisix/upstream.lua index 99e1e857d37e..acb6ab366304 100644 --- a/apisix/upstream.lua +++ b/apisix/upstream.lua @@ -559,6 +559,58 @@ local function get_chash_key_schema(hash_on) end +-- Constraints of `warm_up_conf` within one upstream that JSON schema cannot +-- express. A ramp needs a single roundrobin tier to work in, so a combination it +-- could never act on is rejected at the Admin API rather than accepted and +-- ignored. Where the upstream is used is a different question: like every other +-- field that only applies to HTTP, `warm_up_conf` is simply ignored on the +-- stream path. +-- +-- This runs on the configuration entry points only, never on the data plane +-- checker: a configuration that reaches a running gateway some other way - it is +-- written to etcd or to a standalone config file directly, or it is embedded in a +-- route or a service, neither of which runs this - must not take the whole +-- upstream out of service over a field +-- that only accelerates a ramp. `slow_start.usable()` logs and keeps proxying +-- with the configured weights there. +local function check_warm_up_conf(conf) + local warm_up_conf = conf.warm_up_conf + if not warm_up_conf then + return true + end + + if (conf.type or "roundrobin") ~= "roundrobin" then + return false, "warm_up_conf is only supported by the roundrobin upstream type" + end + + local interval = warm_up_conf.interval or 1 + if interval > warm_up_conf.slow_start_time_seconds then + return false, "warm_up_conf.interval can't be greater than " .. + "warm_up_conf.slow_start_time_seconds" + end + + -- APISIX drains the highest priority tier before it uses the next one, so a + -- weight ramp inside one tier can't hold traffic back from a new node in a + -- tier above the mature ones + local nodes = conf.nodes + if nodes and core.table.isarray(nodes) then + local priority + for i, node in ipairs(nodes) do + local node_priority = node.priority or 0 + if i == 1 then + priority = node_priority + elseif node_priority ~= priority then + return false, "warm_up_conf doesn't support an upstream with " .. + "nodes of different priorities" + end + end + end + + return true +end +_M.check_warm_up_conf = check_warm_up_conf + + local function check_upstream_conf(in_dp, conf) if not in_dp then local ok, err = check_schema(conf) @@ -566,6 +618,11 @@ local function check_upstream_conf(in_dp, conf) return false, "invalid configuration: " .. err end + local ok, err = check_warm_up_conf(conf) + if not ok then + return false, err + end + if conf.nodes and not core.table.isarray(conf.nodes) then local port for addr,_ in pairs(conf.nodes) do diff --git a/conf/config.yaml.example b/conf/config.yaml.example index b201c97fbde0..c18e93171108 100644 --- a/conf/config.yaml.example +++ b/conf/config.yaml.example @@ -355,6 +355,7 @@ nginx_config: # Config for render the template to generate n plugin-limit-conn: 10m worker-events: 10m lrucache-lock: 10m + upstream-slow-start: 10m # Slow start state of upstream nodes (`warm_up_conf`) balancer-ewma: 10m balancer-ewma-locks: 10m balancer-ewma-last-touched-at: 10m diff --git a/docs/en/latest/admin-api.md b/docs/en/latest/admin-api.md index abd9965d3703..77d8d769695b 100644 --- a/docs/en/latest/admin-api.md +++ b/docs/en/latest/admin-api.md @@ -1024,6 +1024,11 @@ In addition to the equalization algorithm selections, Upstream also supports pas | keepalive_pool.size | False | Auxiliary | Sets `keepalive` directive dynamically. | | | keepalive_pool.idle_timeout | False | Auxiliary | Sets `keepalive_timeout` directive dynamically. | | | keepalive_pool.requests | False | Auxiliary | Sets `keepalive_requests` directive dynamically. | | +| warm_up_conf.slow_start_time_seconds | True, when `warm_up_conf` is set | Integer | Slow start window in seconds. A node the gateway observes for the first time takes a reduced share of the traffic and ramps back to its configured weight over this window. Must be at least 1. | 300 | +| warm_up_conf.min_weight_percent | True, when `warm_up_conf` is set | Integer | Lowest effective weight during the ramp, as a percentage of the configured weight, from 1 to 100. | 1 | +| warm_up_conf.interval | False | Integer | Seconds between two effective weight refreshes. Defaults to `1`, and cannot be greater than `slow_start_time_seconds`. | 1 | +| warm_up_conf.aggression | False | Number | Shape of the ramp. `1` (default) is linear, above `1` ramps up faster at the beginning, below `1` slower. At least `0.01`. | 1 | +| warm_up_conf.startup_grace_period_seconds | False | Integer | Seconds after the gateway starts during which a node observed for the first time is treated as already warmed up, so that a restart does not ramp the whole node set again. Defaults to `0`. | 180 | An Upstream can be one of the following `types`: @@ -1067,6 +1072,15 @@ To verify the certificate presented by the Upstream, set `tls.verify` to `true`. To allow Upstream to have a separate connection pool, use `keepalive_pool`. It can be configured by modifying its child fields. +`warm_up_conf` enables slow start for the nodes of a `roundrobin` Upstream. Whether a node is new is decided by the gateway itself, from the node set it observes, and the start of each ramp is recorded locally in the `upstream-slow-start` shared dict: + +- The node set an Upstream has when the gateway first builds a load balancer for it is treated as warmed up. So is the node set it already has when `warm_up_conf` is turned on. +- A node added afterwards ramps from `min_weight_percent` back to its configured weight over `slow_start_time_seconds`. A node held out of the load balancer by a health check starts its ramp when it first becomes available. +- A node that leaves the Upstream and comes back within `slow_start_time_seconds` resumes its ramp. One that comes back later, or that a health check kept out for longer than that, ramps again from the start. +- Every APISIX instance ramps independently, from the moment it observed the node. + +`warm_up_conf` is only supported by `roundrobin` Upstreams whose nodes share a single priority, and it is rejected in the Upstreams of the `traffic-split` Plugin, which are rebuilt per request. Like the other Upstream fields that only apply to HTTP, it is ignored when the Upstream is used by a stream route. A ramp only shifts traffic between nodes: a single-node Upstream, or one whose nodes are all new, keeps sending every request to them. + Example Configuration: ```shell diff --git a/docs/zh/latest/admin-api.md b/docs/zh/latest/admin-api.md index aae200b88f85..ec6aa9c8a7f5 100644 --- a/docs/zh/latest/admin-api.md +++ b/docs/zh/latest/admin-api.md @@ -1032,6 +1032,11 @@ APISIX 的 Upstream 除了基本的负载均衡算法选择外，还支持对上 |keepalive_pool.size | 否 | 辅助 | 动态设置 `keepalive` 指令，详细信息请参考下文。 | |keepalive_pool.idle_timeout | 否 | 辅助 | 动态设置 `keepalive_timeout` 指令，详细信息请参考下文。 | |keepalive_pool.requests | 否 | 辅助 | 动态设置 `keepalive_requests` 指令，详细信息请参考下文。 | +|warm_up_conf.slow_start_time_seconds | 设置 `warm_up_conf` 时必填 | 整型 | 慢启动窗口，单位为秒。网关首次观察到的节点先承接较少的流量，并在该窗口内逐步恢复到配置的权重。最小值为 1。 | +|warm_up_conf.min_weight_percent | 设置 `warm_up_conf` 时必填 | 整型 | 爬坡期间有效权重的下限，以配置权重的百分比表示，取值范围 1 到 100。 | +|warm_up_conf.interval | 否 | 整型 | 两次刷新有效权重之间的间隔，单位为秒。默认为 `1`，不能大于 `slow_start_time_seconds`。 | +|warm_up_conf.aggression | 否 | 数值 | 爬坡曲线。`1`（默认）为线性，大于 `1` 前期增长更快，小于 `1` 前期增长更慢。最小值为 `0.01`。 | +|warm_up_conf.startup_grace_period_seconds | 否 | 整型 | 网关启动后的宽限时间，单位为秒。在此期间首次观察到的节点直接视为已完成预热，避免重启后整组节点重新爬坡。默认为 `0`。 | `type` 详细信息如下： @@ -1070,6 +1075,12 @@ APISIX 的 Upstream 除了基本的负载均衡算法选择外，还支持对上 ``` - `keepalive_pool` 允许 Upstream 有自己单独的连接池。它下属的字段，比如 `requests`，可以用于配置上游连接保持的参数。 +- `warm_up_conf` 为 `roundrobin` 类型的 Upstream 开启节点慢启动。节点是否为新节点由网关根据自己观察到的节点集合判断，每次爬坡的起点记录在本地的 `upstream-slow-start` 共享字典中： + - 网关首次为 Upstream 构建负载均衡器时已有的节点视为已完成预热，开启 `warm_up_conf` 时 Upstream 已有的节点同样如此。 + - 之后新增的节点从 `min_weight_percent` 开始，在 `slow_start_time_seconds` 内逐步恢复到配置的权重。被健康检查挡在负载均衡器之外的节点，从它首次可用时开始爬坡。 + - 节点离开 Upstream 后在 `slow_start_time_seconds` 内回来，会继续原来的爬坡；更晚回来，或被健康检查排除超过该时长，会重新开始爬坡。 + - 每个 APISIX 实例独立计时，起点为该实例观察到节点的时刻。 + - `warm_up_conf` 仅支持节点优先级一致的 `roundrobin` 类型 Upstream，`traffic-split` 插件中的 Upstream 每请求重建，不允许配置。与其他只对 HTTP 生效的 Upstream 字段一样，该字段在 stream route 使用的 Upstream 上会被忽略。爬坡只在节点之间调整流量：单节点 Upstream，或所有节点都是新节点时，请求仍会全部发往这些节点。 Upstream 对象 JSON 配置示例： diff --git a/t/APISIX.pm b/t/APISIX.pm index 76dc94530a92..dca50da5cb04 100644 --- a/t/APISIX.pm +++ b/t/APISIX.pm @@ -660,6 +660,7 @@ _EOC_ lua_shared_dict internal-status 10m; lua_shared_dict worker-events 10m; lua_shared_dict lrucache-lock 10m; + lua_shared_dict upstream-slow-start 10m; lua_shared_dict balancer-ewma 1m; lua_shared_dict balancer-ewma-locks 1m; lua_shared_dict balancer-ewma-last-touched-at 1m; diff --git a/t/admin/upstream-slow-start.t b/t/admin/upstream-slow-start.t new file mode 100644 index 000000000000..58dd861e864c --- /dev/null +++ b/t/admin/upstream-slow-start.t @@ -0,0 +1,622 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# +use t::APISIX 'no_plan'; + +repeat_each(1); +no_long_string(); +no_root_location(); +no_shuffle(); +log_level("info"); + +add_block_preprocessor(sub { + my ($block) = @_; + + if (!$block->request) { + $block->set_value("request", "GET /t"); + } + + if (!$block->no_error_log && !$block->error_log) { + $block->set_value("no_error_log", "[error]\n[alert]"); + } +}); + +run_tests; + +__DATA__ + +=== TEST 1: set upstream with warm_up_conf +--- config + location /t { + content_by_lua_block { + local t = require("lib.test_admin").test + local code, body = t('/apisix/admin/upstreams/1', + ngx.HTTP_PUT, + [[{ + "type": "roundrobin", + "nodes": [ + {"host": "127.0.0.1", "port": 1980, "weight": 100} + ], + "warm_up_conf": { + "slow_start_time_seconds": 300, + "min_weight_percent": 1 + } + }]] + ) + + if code >= 300 then + ngx.status = code + ngx.say(body) + return + end + ngx.say("passed") + } + } +--- response_body +passed + + + +=== TEST 2: defaults are filled in by the schema +--- config + location /t { + content_by_lua_block { + local core = require("apisix.core") + local conf = { + type = "roundrobin", + nodes = {{host = "127.0.0.1", port = 1980, weight = 100}}, + warm_up_conf = { + slow_start_time_seconds = 300, + min_weight_percent = 1, + }, + } + + local ok, err = core.schema.check(core.schema.upstream, conf) + if not ok then + ngx.say("failed: ", err) + return + end + + ngx.say("interval: ", conf.warm_up_conf.interval, + ", aggression: ", conf.warm_up_conf.aggression, + ", startup_grace_period_seconds: ", + conf.warm_up_conf.startup_grace_period_seconds) + } + } +--- response_body +interval: 1, aggression: 1, startup_grace_period_seconds: 0 + + + +=== TEST 3: all fields +--- config + location /t { + content_by_lua_block { + local t = require("lib.test_admin").test + local code, body = t('/apisix/admin/upstreams/1', + ngx.HTTP_PUT, + [[{ + "type": "roundrobin", + "nodes": [ + {"host": "127.0.0.1", "port": 1980, "weight": 100} + ], + "warm_up_conf": { + "slow_start_time_seconds": 300, + "min_weight_percent": 20, + "interval": 5, + "aggression": 2.5, + "startup_grace_period_seconds": 180 + } + }]] + ) + + if code >= 300 then + ngx.status = code + ngx.say(body) + return + end + ngx.say("passed") + } + } +--- response_body +passed + + + +=== TEST 4: slow_start_time_seconds is required +--- config + location /t { + content_by_lua_block { + local t = require("lib.test_admin").test + local code, body = t('/apisix/admin/upstreams/2', + ngx.HTTP_PUT, + [[{ + "type": "roundrobin", + "nodes": [ + {"host": "127.0.0.1", "port": 1980, "weight": 100} + ], + "warm_up_conf": { + "min_weight_percent": 1 + } + }]] + ) + + ngx.status = code + ngx.print(body) + } + } +--- error_code: 400 +--- response_body eval +qr/property .*slow_start_time_seconds.* is required/ + + + +=== TEST 5: min_weight_percent is required +--- config + location /t { + content_by_lua_block { + local t = require("lib.test_admin").test + local code, body = t('/apisix/admin/upstreams/2', + ngx.HTTP_PUT, + [[{ + "type": "roundrobin", + "nodes": [ + {"host": "127.0.0.1", "port": 1980, "weight": 100} + ], + "warm_up_conf": { + "slow_start_time_seconds": 10 + } + }]] + ) + + ngx.status = code + ngx.print(body) + } + } +--- error_code: 400 +--- response_body eval +qr/property .*min_weight_percent.* is required/ + + + +=== TEST 6: min_weight_percent is a percentage, not a ratio +--- config + location /t { + content_by_lua_block { + local t = require("lib.test_admin").test + for _, percent in ipairs({0, 101}) do + local code, body = t('/apisix/admin/upstreams/2', + ngx.HTTP_PUT, + [[{ + "type": "roundrobin", + "nodes": [ + {"host": "127.0.0.1", "port": 1980, "weight": 100} + ], + "warm_up_conf": { + "slow_start_time_seconds": 10, + "min_weight_percent": ]] .. percent .. [[ + } + }]] + ) + if code < 300 then + ngx.say("unexpectedly accepted min_weight_percent ", percent) + return + end + end + ngx.say("passed") + } + } +--- response_body +passed + + + +=== TEST 7: reject slow_start_time_seconds below 1 and aggression below 0.01 +--- config + location /t { + content_by_lua_block { + local t = require("lib.test_admin").test + local cases = { + [["slow_start_time_seconds": 0, "min_weight_percent": 1]], + [["slow_start_time_seconds": 10, "min_weight_percent": 1, "aggression": 0]], + [["slow_start_time_seconds": 10, "min_weight_percent": 1, "interval": 0]], + [["slow_start_time_seconds": 10, "min_weight_percent": 1, "default_weight": 1]], + [["slow_start_time_seconds": 10, "min_weight_percent": 1, + "startup_grace_period_seconds": -1]], + [["slow_start_time_seconds": 10, "min_weight_percent": 1, "unknown": 1]], + } + + for i, case in ipairs(cases) do + local code, body = t('/apisix/admin/upstreams/2', + ngx.HTTP_PUT, + [[{ + "type": "roundrobin", + "nodes": [ + {"host": "127.0.0.1", "port": 1980, "weight": 100} + ], + "warm_up_conf": {]] .. case .. [[} + }]] + ) + if code < 300 then + ngx.say("unexpectedly accepted case ", i) + return + end + end + ngx.say("passed") + } + } +--- response_body +passed + + + +=== TEST 8: reject warm_up_conf on a non roundrobin upstream +--- config + location /t { + content_by_lua_block { + local t = require("lib.test_admin").test + local code, body = t('/apisix/admin/upstreams/2', + ngx.HTTP_PUT, + [[{ + "type": "chash", + "key": "remote_addr", + "nodes": [ + {"host": "127.0.0.1", "port": 1980, "weight": 100} + ], + "warm_up_conf": { + "slow_start_time_seconds": 10, + "min_weight_percent": 1 + } + }]] + ) + + ngx.status = code + ngx.print(body) + } + } +--- error_code: 400 +--- response_body eval +qr/warm_up_conf is only supported by the roundrobin upstream type/ + + + +=== TEST 9: reject an interval longer than the slow start window +--- config + location /t { + content_by_lua_block { + local t = require("lib.test_admin").test + local code, body = t('/apisix/admin/upstreams/2', + ngx.HTTP_PUT, + [[{ + "type": "roundrobin", + "nodes": [ + {"host": "127.0.0.1", "port": 1980, "weight": 100} + ], + "warm_up_conf": { + "slow_start_time_seconds": 10, + "min_weight_percent": 1, + "interval": 11 + } + }]] + ) + + ngx.status = code + ngx.print(body) + } + } +--- error_code: 400 +--- response_body eval +qr/warm_up_conf.interval can't be greater than warm_up_conf.slow_start_time_seconds/ + + + +=== TEST 10: reject nodes with different priorities +--- config + location /t { + content_by_lua_block { + local t = require("lib.test_admin").test + local code, body = t('/apisix/admin/upstreams/2', + ngx.HTTP_PUT, + [[{ + "type": "roundrobin", + "nodes": [ + {"host": "127.0.0.1", "port": 1980, "weight": 100, "priority": 0}, + {"host": "127.0.0.1", "port": 1981, "weight": 100, "priority": -1} + ], + "warm_up_conf": { + "slow_start_time_seconds": 10, + "min_weight_percent": 1 + } + }]] + ) + + ngx.status = code + ngx.print(body) + } + } +--- error_code: 400 +--- response_body eval +qr/warm_up_conf doesn't support an upstream with nodes of different priorities/ + + + +=== TEST 11: accept warm_up_conf on a route embedded upstream +--- config + location /t { + content_by_lua_block { + local t = require("lib.test_admin").test + local code, body = t('/apisix/admin/routes/1', + ngx.HTTP_PUT, + [[{ + "uri": "/hello", + "upstream": { + "type": "roundrobin", + "nodes": [ + {"host": "127.0.0.1", "port": 1980, "weight": 100} + ], + "warm_up_conf": { + "slow_start_time_seconds": 10, + "min_weight_percent": 1 + } + } + }]] + ) + + if code >= 300 then + ngx.status = code + ngx.say(body) + return + end + ngx.say("passed") + } + } +--- response_body +passed + + + +=== TEST 12: accept warm_up_conf on a service embedded upstream +--- config + location /t { + content_by_lua_block { + local t = require("lib.test_admin").test + local code, body = t('/apisix/admin/services/1', + ngx.HTTP_PUT, + [[{ + "upstream": { + "type": "roundrobin", + "nodes": [ + {"host": "127.0.0.1", "port": 1980, "weight": 100} + ], + "warm_up_conf": { + "slow_start_time_seconds": 10, + "min_weight_percent": 1 + } + } + }]] + ) + + if code >= 300 then + ngx.status = code + ngx.say(body) + return + end + ngx.say("passed") + } + } +--- response_body +passed + + + +=== TEST 13: reject warm_up_conf on a route embedded upstream that is not roundrobin +--- config + location /t { + content_by_lua_block { + local t = require("lib.test_admin").test + local code, body = t('/apisix/admin/routes/2', + ngx.HTTP_PUT, + [[{ + "uri": "/hello", + "upstream": { + "type": "least_conn", + "nodes": [ + {"host": "127.0.0.1", "port": 1980, "weight": 100} + ], + "warm_up_conf": { + "slow_start_time_seconds": 10, + "min_weight_percent": 1 + } + } + }]] + ) + + ngx.status = code + ngx.print(body) + } + } +--- error_code: 400 +--- response_body eval +qr/warm_up_conf is only supported by the roundrobin upstream type/ + + + +=== TEST 14: reject warm_up_conf in a traffic-split upstream +--- config + location /t { + content_by_lua_block { + local t = require("lib.test_admin").test + local code, body = t('/apisix/admin/routes/3', + ngx.HTTP_PUT, + [[{ + "uri": "/hello", + "plugins": { + "traffic-split": { + "rules": [{ + "weighted_upstreams": [{ + "upstream": { + "type": "roundrobin", + "nodes": [ + {"host": "127.0.0.1", "port": 1981, "weight": 100} + ], + "warm_up_conf": { + "slow_start_time_seconds": 10, + "min_weight_percent": 1 + } + }, + "weight": 1 + }] + }] + } + }, + "upstream": { + "type": "roundrobin", + "nodes": [ + {"host": "127.0.0.1", "port": 1980, "weight": 100} + ] + } + }]] + ) + + ngx.status = code + ngx.print(body) + } + } +--- error_code: 400 +--- response_body eval +qr/warm_up_conf is not supported by the upstream of the traffic-split plugin/ + + + +=== TEST 15: declarative validation accepts warm_up_conf +--- config + location /t { + content_by_lua_block { + local t = require("lib.test_admin").test + local code, body = t('/apisix/admin/configs/validate', + ngx.HTTP_POST, + [[{ + "upstreams": [ + { + "id": "u1", + "type": "roundrobin", + "nodes": {"127.0.0.1:1980": 1}, + "warm_up_conf": { + "slow_start_time_seconds": 10, + "min_weight_percent": 1 + } + } + ] + }]] + ) + + ngx.status = code + ngx.say(body) + } + } +--- error_code: 200 +--- response_body +passed + + + +=== TEST 16: declarative validation still rejects an unusable warm_up_conf +--- config + location /t { + content_by_lua_block { + local t = require("lib.test_admin").test + local code, body = t('/apisix/admin/configs/validate', + ngx.HTTP_POST, + [[{ + "upstreams": [ + { + "id": "u1", + "type": "chash", + "key": "remote_addr", + "nodes": {"127.0.0.1:1980": 1}, + "warm_up_conf": { + "slow_start_time_seconds": 10, + "min_weight_percent": 1 + } + } + ] + }]] + ) + + ngx.status = code + ngx.print(body) + } + } +--- error_code: 400 +--- response_body eval +qr/warm_up_conf is only supported by the roundrobin upstream type/ + + + +=== TEST 17: the data plane keeps an upstream it cannot ramp +--- config + location /t { + content_by_lua_block { + local core = require("apisix.core") + local apisix_upstream = require("apisix.upstream") + + -- the Admin API rejects this, but a control plane or an embedded + -- upstream reaches the data plane without passing through it; the + -- upstream still has to load, or every route using it returns 503 + local conf = { + type = "chash", + key = "remote_addr", + nodes = {{host = "127.0.0.1", port = 1980, weight = 100, priority = 0}}, + warm_up_conf = { + slow_start_time_seconds = 10, + min_weight_percent = 1, + }, + } + + local ok, err = apisix_upstream.check_upstream_conf(conf) + ngx.say("admin: ", tostring(ok), " ", tostring(err)) + + local dp_conf = core.table.deepcopy(conf) + local dp_ok, dp_err = core.schema.check(core.schema.upstream, dp_conf) + ngx.say("data plane schema: ", tostring(dp_ok), " ", tostring(dp_err)) + } + } +--- response_body +admin: false warm_up_conf is only supported by the roundrobin upstream type +data plane schema: true nil + + + +=== TEST 18: clean up +--- config + location /t { + content_by_lua_block { + local t = require("lib.test_admin").test + for _, uri in ipairs({'/apisix/admin/routes/1', + '/apisix/admin/services/1', + '/apisix/admin/upstreams/1?force=true'}) do + local code, body = t(uri, ngx.HTTP_DELETE) + if code >= 300 then + ngx.status = code + ngx.say(uri, ": ", body) + return + end + end + ngx.say("passed") + } + } +--- response_body +passed diff --git a/t/node/upstream-slow-start.t b/t/node/upstream-slow-start.t new file mode 100644 index 000000000000..4fa385b4735c --- /dev/null +++ b/t/node/upstream-slow-start.t @@ -0,0 +1,833 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# +use t::APISIX 'no_plan'; + +repeat_each(1); +no_long_string(); +no_root_location(); +no_shuffle(); +log_level("info"); +workers(1); + +add_block_preprocessor(sub { + my ($block) = @_; + + if (!$block->request) { + $block->set_value("request", "GET /t"); + } + + if (!$block->no_error_log && !$block->error_log) { + $block->set_value("no_error_log", "[error]\n[alert]"); + } +}); + +run_tests; + +__DATA__ + +=== TEST 1: the effective weight follows the ramp +--- config + location /t { + content_by_lua_block { + local slow_start = require("apisix.slow_start") + local conf = { + slow_start_time_seconds = 300, + min_weight_percent = 1, + interval = 1, + aggression = 1, + } + local now = 10000 + + -- the first bucket counts as one second of progress, not zero: over a + -- 10s window that is a tenth of the weight, well above the 1% floor + ngx.say(slow_start.effective_weight(100, now, { + slow_start_time_seconds = 10, + min_weight_percent = 1, + aggression = 1, + }, now)) + ngx.say(slow_start.effective_weight(100, now, conf, now)) + ngx.say(slow_start.effective_weight(100, now, conf, now + 30)) + ngx.say(slow_start.effective_weight(100, now, conf, now + 150)) + ngx.say(slow_start.effective_weight(100, now, conf, now + 300)) + ngx.say(slow_start.effective_weight(100, now, conf, now + 3000)) + -- a node configured out of the rotation stays out of it + ngx.say(slow_start.effective_weight(0, now, conf, now + 30)) + -- anything above zero keeps at least one weight unit + ngx.say(slow_start.effective_weight(1, now, conf, now + 1)) + -- a node with no state, and a node marked mature, use the full weight + ngx.say(slow_start.effective_weight(100, nil, conf, now + 1)) + ngx.say(slow_start.effective_weight(100, 0, conf, now + 1)) + } + } +--- response_body +10 +1 +10 +50 +100 +100 +0 +1 +100 +100 + + + +=== TEST 2: min_weight_percent is the floor, aggression the shape +--- config + location /t { + content_by_lua_block { + local slow_start = require("apisix.slow_start") + local now = 10000 + local function weight(percent, aggression, elapsed) + return slow_start.effective_weight(100, now, { + slow_start_time_seconds = 300, + min_weight_percent = percent, + aggression = aggression, + }, now + elapsed) + end + + ngx.say(weight(20, 1, 1)) + ngx.say(weight(1, 1, 30)) + ngx.say(weight(1, 2, 30) > weight(1, 1, 30)) + ngx.say(weight(1, 0.5, 30) < weight(1, 1, 30)) + } + } +--- response_body +20 +10 +true +true + + + +=== TEST 3: a local clock jump does not produce a weight outside the range +--- config + location /t { + content_by_lua_block { + local slow_start = require("apisix.slow_start") + local conf = { + slow_start_time_seconds = 300, + min_weight_percent = 1, + aggression = 1, + } + + -- backwards + ngx.say(slow_start.effective_weight(100, 10000, conf, 9000)) + -- and forwards, clamped to the full window + ngx.say(slow_start.effective_weight(100, 10000, conf, 99999)) + } + } +--- response_body +1 +100 + + + +=== TEST 4: a node added to a running upstream ramps up, the existing one does not +--- timeout: 30 +--- config + location /t { + content_by_lua_block { + local t = require("lib.test_admin").test + local http = require("resty.http") + local uri = "http://127.0.0.1:" .. ngx.var.server_port .. "/server_port" + + local function set_upstream(nodes, desc) + return t('/apisix/admin/upstreams/1', + ngx.HTTP_PUT, + [[{ + "type": "roundrobin", + "desc": "]] .. desc .. [[", + "nodes": ]] .. nodes .. [[, + "warm_up_conf": { + "slow_start_time_seconds": 10, + "min_weight_percent": 1, + "interval": 1 + } + }]] + ) + end + + local function split(times) + local ports = {} + for _ = 1, times do + local httpc = http.new() + local res, err = httpc:request_uri(uri) + if not res then + return nil, err + end + ports[res.body] = (ports[res.body] or 0) + 1 + end + return ports + end + + local one = [=[[{"host": "127.0.0.1", "port": 1980, "weight": 100}]]=] + local two = [=[[{"host": "127.0.0.1", "port": 1980, "weight": 100}, + {"host": "127.0.0.1", "port": 1981, "weight": 100}]]=] + + local code, body = set_upstream(one, "one node") + if code >= 300 then + ngx.status = code + ngx.say(body) + return + end + + code, body = t('/apisix/admin/routes/1', + ngx.HTTP_PUT, + [[{ + "uri": "/server_port", + "upstream_id": "1" + }]] + ) + if code >= 300 then + ngx.status = code + ngx.say(body) + return + end + + -- the first picker build records the bootstrapped node as the baseline + ngx.sleep(0.5) + split(1) + + code, body = set_upstream(two, "two nodes") + if code >= 300 then + ngx.status = code + ngx.say(body) + return + end + ngx.sleep(0.5) + + local ports, err = split(20) + if not ports then + ngx.say(err) + return + end + local mature, warming = ports["1980"] or 0, ports["1981"] or 0 + ngx.log(ngx.WARN, "ramping split: 1980=", mature, " 1981=", warming) + if not (mature >= 17 and warming <= 3) then + ngx.say("failed while ramping: 1980=", mature, " 1981=", warming) + return + end + + -- once the window has passed both nodes are back to their weights + ngx.sleep(11) + ports, err = split(20) + if not ports then + ngx.say(err) + return + end + if ports["1980"] ~= 10 or ports["1981"] ~= 10 then + ngx.say("failed after the window: 1980=", tostring(ports["1980"]), + " 1981=", tostring(ports["1981"])) + return + end + + -- an unrelated field changing must not start the ramp over + code, body = set_upstream(two, "a new description") + if code >= 300 then + ngx.status = code + ngx.say(body) + return + end + ngx.sleep(0.5) + + ports, err = split(20) + if not ports then + ngx.say(err) + return + end + if ports["1980"] ~= 10 or ports["1981"] ~= 10 then + ngx.say("failed after an unrelated change: 1980=", + tostring(ports["1980"]), " 1981=", tostring(ports["1981"])) + return + end + + ngx.say("passed") + } + } +--- response_body +passed +--- error_log +slow start began for node 127.0.0.1:1981 +slow start finished for node 127.0.0.1:1981 +--- no_error_log eval +[qr/\[error\]/, qr/\[alert\]/, qr/slow start began for node 127\.0\.0\.1:1980/] + + + +=== TEST 5: a node that comes back inside the tombstone window keeps its progress +--- timeout: 30 +--- config + location /t { + content_by_lua_block { + local t = require("lib.test_admin").test + local http = require("resty.http") + local uri = "http://127.0.0.1:" .. ngx.var.server_port .. "/slow-start-tombstone" + + local function set_upstream(nodes) + return t('/apisix/admin/upstreams/2', + ngx.HTTP_PUT, + [[{ + "type": "roundrobin", + "nodes": ]] .. nodes .. [[, + "warm_up_conf": { + "slow_start_time_seconds": 10, + "min_weight_percent": 1, + "interval": 10 + } + }]] + ) + end + + local function split(times) + local ports = {} + for _ = 1, times do + local httpc = http.new() + local res, err = httpc:request_uri(uri) + if not res then + return nil, err + end + ports[res.body] = (ports[res.body] or 0) + 1 + end + return ports + end + + local one = [=[[{"host": "127.0.0.1", "port": 1980, "weight": 100}]]=] + local two = [=[[{"host": "127.0.0.1", "port": 1980, "weight": 100}, + {"host": "127.0.0.1", "port": 1981, "weight": 100}]]=] + + local code, body = set_upstream(one) + if code >= 300 then + ngx.status = code + ngx.say(body) + return + end + + code, body = t('/apisix/admin/routes/2', + ngx.HTTP_PUT, + [[{ + "uri": "/slow-start-tombstone", + "plugins": { + "proxy-rewrite": {"uri": "/server_port"} + }, + "upstream_id": "2" + }]] + ) + if code >= 300 then + ngx.status = code + ngx.say(body) + return + end + + ngx.sleep(0.5) + split(1) + + -- 1981 joins and starts its ramp + set_upstream(two) + ngx.sleep(0.5) + split(1) + + -- it leaves, and comes back well inside the 10s tombstone window + set_upstream(one) + ngx.sleep(0.5) + split(1) + ngx.sleep(4) + set_upstream(two) + ngx.sleep(0.5) + + local ports, err = split(100) + if not ports then + ngx.say(err) + return + end + + -- the ramp continued while it was away, so 1981 is about half way + -- through the window: a real share of the traffic, still below 1980. + -- Restarting the ramp would have put it back at a tenth of the weight + local mature, warming = ports["1980"] or 0, ports["1981"] or 0 + ngx.log(ngx.WARN, "tombstone split: 1980=", mature, " 1981=", warming) + if warming >= 15 and warming < mature then + ngx.say("passed") + else + ngx.say("failed: 1980=", mature, " 1981=", warming) + end + } + } +--- response_body +passed +--- grep_error_log eval +qr/slow start began for node \S+/ +--- grep_error_log_out +slow start began for node 127.0.0.1:1981 + + + +=== TEST 6: without warm_up_conf the weights are used as configured +--- config + location /t { + content_by_lua_block { + local t = require("lib.test_admin").test + local code, body = t('/apisix/admin/upstreams/3', + ngx.HTTP_PUT, + [[{ + "type": "roundrobin", + "nodes": [ + {"host": "127.0.0.1", "port": 1980, "weight": 100}, + {"host": "127.0.0.1", "port": 1981, "weight": 100} + ] + }]] + ) + if code >= 300 then + ngx.status = code + ngx.say(body) + return + end + + code, body = t('/apisix/admin/routes/3', + ngx.HTTP_PUT, + [[{ + "uri": "/slow-start-off", + "plugins": { + "proxy-rewrite": {"uri": "/server_port"} + }, + "upstream_id": "3" + }]] + ) + if code >= 300 then + ngx.status = code + ngx.say(body) + return + end + + ngx.sleep(0.5) + + local http = require("resty.http") + local uri = "http://127.0.0.1:" .. ngx.var.server_port .. "/slow-start-off" + local ports = {} + for _ = 1, 20 do + local httpc = http.new() + local res, err = httpc:request_uri(uri) + if not res then + ngx.say(err) + return + end + ports[res.body] = (ports[res.body] or 0) + 1 + end + + if ports["1980"] == 10 and ports["1981"] == 10 then + ngx.say("passed") + else + ngx.say("failed: 1980=", tostring(ports["1980"]), + " 1981=", tostring(ports["1981"])) + end + } + } +--- response_body +passed +--- no_error_log +slow start began for node + + + +=== TEST 7: a route embedded upstream is its own lifecycle scope +--- timeout: 15 +--- config + location /t { + content_by_lua_block { + local t = require("lib.test_admin").test + local http = require("resty.http") + local uri = "http://127.0.0.1:" .. ngx.var.server_port .. "/slow-start-embedded" + + local function set_route(nodes) + return t('/apisix/admin/routes/4', + ngx.HTTP_PUT, + [[{ + "uri": "/slow-start-embedded", + "plugins": { + "proxy-rewrite": {"uri": "/server_port"} + }, + "upstream": { + "type": "roundrobin", + "nodes": ]] .. nodes .. [[, + "warm_up_conf": { + "slow_start_time_seconds": 10, + "min_weight_percent": 1, + "interval": 1 + } + } + }]] + ) + end + + local one = [=[[{"host": "127.0.0.1", "port": 1980, "weight": 100}]]=] + local two = [=[[{"host": "127.0.0.1", "port": 1980, "weight": 100}, + {"host": "127.0.0.1", "port": 1981, "weight": 100}]]=] + + local code, body = set_route(one) + if code >= 300 then + ngx.status = code + ngx.say(body) + return + end + + ngx.sleep(0.5) + local httpc = http.new() + httpc:request_uri(uri) + + code, body = set_route(two) + if code >= 300 then + ngx.status = code + ngx.say(body) + return + end + ngx.sleep(0.5) + + local ports = {} + for _ = 1, 20 do + httpc = http.new() + local res, err = httpc:request_uri(uri) + if not res then + ngx.say(err) + return + end + ports[res.body] = (ports[res.body] or 0) + 1 + end + + local mature, warming = ports["1980"] or 0, ports["1981"] or 0 + if mature >= 17 and warming <= 3 then + ngx.say("passed") + else + ngx.say("failed: 1980=", mature, " 1981=", warming) + end + } + } +--- response_body +passed +--- error_log eval +qr{of upstream \S*/routes/4} + + + +=== TEST 8: the picker cache key changes once per interval, then settles +--- config + location /t { + content_by_lua_block { + local core = require("apisix.core") + local slow_start = require("apisix.slow_start") + local dict = ngx.shared["upstream-slow-start"] + local up_conf = { + resource_key = "/upstreams/version-suffix", + type = "roundrobin", + warm_up_conf = { + slow_start_time_seconds = 10, + min_weight_percent = 1, + interval = 2, + }, + } + + -- no deadline published yet: the build this triggers is the one that + -- publishes it, so the key already carries a bucket + local first = slow_start.version_suffix(up_conf) + ngx.say("bucketed: ", first ~= nil and first:match("^#w%d+$") ~= nil) + + -- a second request inside the same bucket reuses the cached picker + ngx.say("stable within the interval: ", + slow_start.version_suffix(up_conf) == first) + + local function publish_deadline(deadline) + local ok, err = dict:set(up_conf.resource_key .. "|!", deadline) + if not ok then + error("failed to publish the deadline: " .. err) + end + end + + -- while a node ramps, crossing the bucket rebuilds it exactly once + publish_deadline(ngx.now() + 10) + ngx.sleep(2.1) + local next_bucket = slow_start.version_suffix(up_conf) + ngx.say("rebuilt after the interval: ", next_bucket ~= first) + ngx.say("stable again: ", slow_start.version_suffix(up_conf) == next_bucket) + + -- once every node is mature the key stops moving altogether + publish_deadline(0) + ngx.say("settled: ", slow_start.version_suffix(up_conf)) + ngx.sleep(2.1) + ngx.say("still settled: ", slow_start.version_suffix(up_conf)) + + dict:delete(up_conf.resource_key .. "|!") + } + } +--- timeout: 15 +--- response_body +bucketed: true +stable within the interval: true +rebuilt after the interval: true +stable again: true +settled: #wm +still settled: #wm + + + +=== TEST 9: an unhealthy node keeps its lifecycle, it is not treated as removed +--- timeout: 20 +--- config + location /t { + content_by_lua_block { + local t = require("lib.test_admin").test + local http = require("resty.http") + + -- 1979 has nothing listening, so the active check takes it out of the + -- picker while the configuration still holds it + local code, body = t('/apisix/admin/upstreams/5', + ngx.HTTP_PUT, + [[{ + "type": "roundrobin", + "nodes": [ + {"host": "127.0.0.1", "port": 1980, "weight": 100}, + {"host": "127.0.0.1", "port": 1979, "weight": 100} + ], + "checks": { + "active": { + "http_path": "/status", + "healthy": {"interval": 1, "successes": 1}, + "unhealthy": {"interval": 1, "tcp_failures": 1} + } + }, + "warm_up_conf": { + "slow_start_time_seconds": 10, + "min_weight_percent": 1, + "interval": 1 + } + }]] + ) + if code >= 300 then + ngx.status = code + ngx.say(body) + return + end + + code, body = t('/apisix/admin/routes/5', + ngx.HTTP_PUT, + [[{ + "uri": "/slow-start-unhealthy", + "plugins": { + "proxy-rewrite": {"uri": "/server_port"} + }, + "upstream_id": "5" + }]] + ) + if code >= 300 then + ngx.status = code + ngx.say(body) + return + end + + local uri = "http://127.0.0.1:" .. ngx.var.server_port .. "/slow-start-unhealthy" + local function hit(times) + local ports = {} + for _ = 1, times do + local httpc = http.new() + local res, err = httpc:request_uri(uri) + if not res then + return nil, err + end + ports[res.body] = (ports[res.body] or 0) + 1 + end + return ports + end + + -- requests before the checker has taken 1979 out may land on it and + -- come back as a 502; only a request that cannot be made at all fails + local ports, err + ngx.sleep(0.5) + ports, err = hit(2) + if not ports then + ngx.say("request failed: ", err) + return + end + -- let the active checker settle and keep rebuilding the picker + ngx.sleep(3) + ports, err = hit(5) + if not ports then + ngx.say("request failed: ", err) + return + end + ngx.sleep(2) + ports, err = hit(10) + if not ports then + ngx.say("request failed: ", err) + return + end + + ngx.say("1980: ", ports["1980"] or 0) + } + } +--- response_body +1980: 10 +--- no_error_log +left the upstream + + + +=== TEST 10: a node kept out of the picker starts its window when it first gets in +--- config + location /t { + content_by_lua_block { + local slow_start = require("apisix.slow_start") + local n1 = {host = "10.0.0.1", port = 8080, weight = 100, priority = 0} + local n2 = {host = "10.0.0.2", port = 8080, weight = 100, priority = 0} + local up_conf = { + resource_key = "/upstreams/pending", + type = "roundrobin", + warm_up_conf = { + slow_start_time_seconds = 100, + min_weight_percent = 1, + interval = 1, + aggression = 1, + }, + } + + -- the upstream is bootstrapped with one node + up_conf.nodes = {n1} + local weights = slow_start.effective_weights(up_conf, {n1}) + ngx.say("baseline: ", weights[1]) + + -- a second node is configured, but a health check keeps it out of the + -- picker for a while. It must not age into maturity while sidelined + up_conf.nodes = {n1, n2} + weights = slow_start.effective_weights(up_conf, {n1}) + ngx.say("while sidelined: ", weights[1], " ", tostring(weights[2])) + + -- the first picker it reaches is where its window starts + weights = slow_start.effective_weights(up_conf, {n1, n2}) + ngx.say("first picker: ", weights[1], " ", weights[2]) + } + } +--- response_body +baseline: 100 +while sidelined: 100 nil +first picker: 100 1 +--- error_log +slow start began for node 10.0.0.2:8080 + + + +=== TEST 11: a worker never writes back a ramp start another worker just changed +--- config + location /t { + content_by_lua_block { + local slow_start = require("apisix.slow_start") + local dict = ngx.shared["upstream-slow-start"] + local mt = getmetatable(dict) + local orig_get = mt.get + + -- Reconciles in different workers run truly in parallel. Replay the + -- losing interleaving deterministically: this worker reads a node's + -- state, and another worker writes before this one does + local function race(target, other_worker) + local fired = false + mt.get = function(self, key, ...) + local value, flags = orig_get(self, key, ...) + if self == dict and key == target and not fired then + fired = true + other_worker() + end + return value, flags + end + end + + local function run(fn) + local ok, err = pcall(fn) + mt.get = orig_get + if not ok then + error(err) + end + end + + local conf = { + slow_start_time_seconds = 100, + min_weight_percent = 1, + interval = 1, + aggression = 1, + } + local n1 = {host = "10.0.1.1", port = 80, weight = 100, priority = 0} + local n2 = {host = "10.0.1.2", port = 80, weight = 100, priority = 0} + local scope = "/upstreams/race" + local key = scope .. "|10.0.1.2:80" + local up_conf = {resource_key = scope, type = "roundrobin", warm_up_conf = conf} + + up_conf.nodes = {n1} + slow_start.effective_weights(up_conf, {n1}) + up_conf.nodes = {n1, n2} + slow_start.effective_weights(up_conf, {n1}) + ngx.say("pending: ", dict:get(key)) + + -- both workers see n2 enter the picker; the other one elects first + local elected = ngx.now() - 50 + local weights + run(function() + race(key, function() + assert(dict:add(key .. "@-1", elected, 10)) + assert(dict:set(key, elected)) + end) + weights = slow_start.effective_weights(up_conf, {n1, n2}) + end) + ngx.say("adopted the elected start: ", dict:get(key) == elected, + ", weight ", weights[2]) + + -- the other worker finishes the ramp while this one, having read the + -- old start, only means to refresh it + run(function() + race(key, function() + assert(dict:add(key .. "@" .. elected, 0, 10)) + assert(dict:set(key, 0)) + end) + slow_start.effective_weights(up_conf, {n1, n2}) + end) + ngx.say("refresh left the change alone: ", dict:get(key)) + } + } +--- response_body +pending: -1 +adopted the elected start: true, weight 50 +refresh left the change alone: 0 + + + +=== TEST 12: clean up +--- config + location /t { + content_by_lua_block { + local t = require("lib.test_admin").test + for _, uri in ipairs({'/apisix/admin/routes/1', '/apisix/admin/routes/2', + '/apisix/admin/routes/3', '/apisix/admin/routes/4', + '/apisix/admin/routes/5', + '/apisix/admin/upstreams/1?force=true', '/apisix/admin/upstreams/2?force=true', + '/apisix/admin/upstreams/3?force=true', '/apisix/admin/upstreams/5?force=true'}) do + local code, body = t(uri, ngx.HTTP_DELETE) + if code >= 300 then + ngx.status = code + ngx.say(uri, ": ", body) + return + end + end + ngx.say("passed") + } + } +--- response_body +passed