feat(upstream): slow start for newly observed upstream nodes

apisix/admin/stream_routes.lua

@@ -23,27 +23,81 @@ local ipairs = ipairs
 local type = type
 
 
+-- etcd hands a resource back either already decoded or as the raw JSON text.
+-- A decode failure and a JSON `null` both have to be rejected here: `null`
+-- decodes to the truthy `core.json.null` userdata, which blows up on the first
+-- field access instead of failing validation.
+local function decode_value(kind, id, value)
+    if type(value) == "table" then
+        return value
+    end
+
+    if type(value) ~= "string" then
+        return nil, {error_msg = "failed to read " .. kind .. " [" .. id .. "]: "
+                                 .. "unexpected value type " .. type(value)}
+    end
+
+    local decoded, decode_err = core.json.decode(value)
+    if type(decoded) ~= "table" then
+        return nil, {error_msg = "failed to decode " .. kind .. " [" .. id .. "]: "
+                                 .. (decode_err or "not an object")}
+    end
+
+    return decoded
+end
+
+
+-- Slow start only ramps HTTP upstreams, so an upstream a stream route can reach
+-- may not enable it. The route reaches one directly through `upstream_id`, or
+-- through a service that embeds one or names one of its own.
+local function check_upstream_reference(upstream_id, via)
+    local key = "/upstreams/" .. upstream_id
+    local res, err = core.etcd.get(key)
+    if not res then
+        return nil, {error_msg = "failed to fetch upstream info by "
+                                 .. "upstream id [" .. upstream_id .. "]: " .. err}
+    end
+
+    if res.status ~= 200 then
+        return nil, {error_msg = "failed to fetch upstream info by "
+                                 .. "upstream id [" .. upstream_id .. "], "
+                                 .. "response code: " .. res.status}
+    end
+
+    local upstream, decode_err = decode_value("upstream", upstream_id,
+                                              res.body.node and res.body.node.value)
+    if not upstream then
+        return nil, decode_err
+    end
+
+    if upstream.warm_up_conf then
+        return nil, {error_msg = (via or ("upstream [" .. upstream_id .. "]"))
+                                 .. " uses warm_up_conf, which is not supported by "
+                                 .. "a stream route"}
+    end
+
+    return true
+end
+
+
 local function check_conf(id, conf, need_id, schema, opts)
     opts = opts or {}
     local ok, err = core.schema.check(schema, conf)
     if not ok then
         return nil, {error_msg = "invalid configuration: " .. err}
     end
 
+    -- slow start only ramps HTTP upstreams, so a stream route may neither carry
+    -- nor point at an upstream that asks for it
+    if conf.upstream and conf.upstream.warm_up_conf then
+        return nil, {error_msg = "warm_up_conf is not supported by a stream route"}
+    end
+
     local upstream_id = conf.upstream_id
     if upstream_id and not opts.skip_references_check then
-        local key = "/upstreams/" .. upstream_id
-        local res, err = core.etcd.get(key)
-        if not res then
-            return nil, {error_msg = "failed to fetch upstream info by "
-                                     .. "upstream id [" .. upstream_id .. "]: "
-                                     .. err}
-        end
-
-        if res.status ~= 200 then
-            return nil, {error_msg = "failed to fetch upstream info by "
-                                     .. "upstream id [" .. upstream_id .. "], "
-                                     .. "response code: " .. res.status}
+        local ok, err = check_upstream_reference(upstream_id)
+        if not ok then
+            return nil, err
         end
     end
 
@@ -62,6 +116,34 @@ local function check_conf(id, conf, need_id, schema, opts)
                     .. "service id [" .. service_id .. "], "
                     .. "response code: " .. res.status}
         end
+
+        -- a service reaches the same upstream, so it can carry warm_up_conf onto
+        -- the L4 path the same way a directly referenced upstream would. The
+        -- route only falls back to the service's upstream when it names none of
+        -- its own, which is what `merge_service_stream_route` does at runtime
+        local service, decode_err = decode_value("service", service_id,
+                                                 res.body.node and res.body.node.value)
+        if not service then
+            return nil, decode_err
+        end
+
+        if not upstream_id then
+            if service.upstream and service.upstream.warm_up_conf then
+                return nil, {error_msg = "service [" .. service_id .. "] uses an "
+                                         .. "upstream with warm_up_conf, which is not "
+                                         .. "supported by a stream route"}
+            end
+
+            if service.upstream_id then
+                local ok, err = check_upstream_reference(service.upstream_id,
+                                                         "service [" .. service_id
+                                                         .. "] upstream ["
+                                                         .. service.upstream_id .. "]")
+                if not ok then
+                    return nil, err
+                end
+            end
+        end
     end
 
     -- the self-reference check needs no lookup, so it stays outside the gate;
@@ -87,17 +169,13 @@ local function check_conf(id, conf, need_id, schema, opts)
                                      .. "], response code: " .. res.status}
         end
 
-        local superior_route = res.body.node.value
-        if type(superior_route) == "string" then
-            local decoded, decode_err = core.json.decode(superior_route)
-            if not decoded then
-                return nil, {error_msg = "failed to decode stream routes[" .. superior_id
-                                         .. "]: " .. decode_err}
-            end
-            superior_route = decoded
+        local superior_route, decode_err = decode_value("stream route", superior_id,
+                                                        res.body.node and res.body.node.value)
+        if not superior_route then
+            return nil, decode_err
         end
 
-        if superior_route and superior_route.protocol
+        if superior_route.protocol
            and superior_route.protocol.name ~= conf.protocol.name then
             return nil, {error_msg = "protocol mismatch: subordinate protocol ["
                                      .. conf.protocol.name .. "] does not match superior protocol ["

apisix/admin/upstreams.lua

@@ -26,14 +26,114 @@ local apisix_upstream = require("apisix.upstream")
 local resource = require("apisix.admin.resource")
 local tostring = tostring
 local ipairs = ipairs
+local type = type
 
 
-local function check_conf(id, conf, need_id)
+local function list_resources(path)
+    local res, err = core.etcd.get(path, true)
+    if not res then
+        return nil, {error_msg = "failed to fetch " .. path .. ": " .. err}
+    end
+
+    -- a prefix nothing has been written under yet is a 404, not an error
+    if res.status == 404 then
+        return {}
+    end
+
+    if res.status ~= 200 then
+        return nil, {error_msg = "failed to fetch " .. path .. ", response code: "
+                                 .. res.status}
+    end
+
+    local nodes = res.body.list
+    if not nodes and res.body.node then
+        nodes = res.body.node.nodes
+    end
+
+    local values = {}
+    for _, item in ipairs(nodes or {}) do
+        local value = item.value
+        if type(value) == "string" then
+            value = core.json.decode(value)
+        end
+
+        if type(value) == "table" then
+            core.table.insert(values, value)
+        end
+    end
+
+    return values
+end
+
+
+-- The stream subsystem never ramps node weights, so an upstream a stream route
+-- can reach may not enable slow start: the configuration would be accepted and
+-- then silently ignored on the L4 path. A route reaches one through its own
+-- `upstream_id`, or - when it names none - through the service it uses, which is
+-- the fallback `merge_service_stream_route` applies at runtime.
+local function check_stream_route_reference(id, conf, opts)
+    if not (conf.warm_up_conf and id) or opts.skip_references_check then
+        return true
+    end
+
+    local routes, err = list_resources("/stream_routes")
+    if not routes then
+        return nil, err
+    end
+
+    local via_service = {}
+    local has_service_ref = false
+    for _, route in ipairs(routes) do
+        if route.upstream_id and tostring(route.upstream_id) == tostring(id) then
+            return nil, {error_msg = "can not enable warm_up_conf on this upstream, "
+                                     .. "stream route [" .. tostring(route.id)
+                                     .. "] is using it now"}
+        end
+
+        if route.service_id and not route.upstream_id then
+            via_service[tostring(route.service_id)] = tostring(route.id)
+            has_service_ref = true
+        end
+    end
+
+    if not has_service_ref then
+        return true
+    end
+
+    local services, err = list_resources("/services")
+    if not services then
+        return nil, err
+    end
+
+    for _, service in ipairs(services) do
+        local route_id = via_service[tostring(service.id)]
+        if route_id and service.upstream_id
+           and tostring(service.upstream_id) == tostring(id) then
+
+            return nil, {error_msg = "can not enable warm_up_conf on this upstream, "
+                                     .. "stream route [" .. route_id .. "] is using it "
+                                     .. "through service [" .. tostring(service.id)
+                                     .. "] now"}
+        end
+    end
+
+    return true
+end
+
+
+local function check_conf(id, conf, need_id, schema, opts)
+    opts = opts or {}
+
     local ok, err = apisix_upstream.check_upstream_conf(conf)
     if not ok then
         return nil, {error_msg = err}
     end
 
+    local ok, err = check_stream_route_reference(id, conf, opts)
+    if not ok then
+        return nil, err
+    end
+
     return true
 end
 

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

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",

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"] *};