This is an automated email from the ASF dual-hosted git repository. shreemaan-abhishek pushed a commit to branch backport/14167-release-3.20 in repository https://gitbox.apache.org/repos/asf/apisix.git
commit 1b673c272346379ecd5e591a5dd0de4bf261e42b Author: Ming Wen <[email protected]> AuthorDate: Fri Nov 6 16:02:48 2026 +0800 fix(discovery): refresh the kubernetes endpoint cache by ttl instead of a version key (#14167) (cherry picked from commit 22ecbc60e4ea780abc0c76ee6396583a26ff7e68) --- apisix/discovery/kubernetes/core.lua | 28 +-- apisix/discovery/kubernetes/init.lua | 37 ++- apisix/discovery/tars/init.lua | 32 +-- docs/en/latest/discovery/kubernetes.md | 2 +- docs/zh/latest/discovery/kubernetes.md | 2 +- t/kubernetes/discovery/kubernetes2.t | 2 +- t/kubernetes/discovery/kubernetes3.t | 8 +- t/kubernetes/discovery/kubernetes4.t | 4 +- t/kubernetes/discovery/kubernetes5.t | 33 ++- t/kubernetes/discovery/kubernetes6.t | 372 +++++++++++++++++++++++++++++ t/kubernetes/discovery/kubernetes7.t | 135 +++++++++++ t/kubernetes/discovery/stream/kubernetes.t | 2 +- t/lib/k8s_discovery_e2e.lua | 235 ++++++++++++++++++ t/tars/discovery/tars.t | 20 ++ 14 files changed, 832 insertions(+), 80 deletions(-) diff --git a/apisix/discovery/kubernetes/core.lua b/apisix/discovery/kubernetes/core.lua index aaf0142a8..460bf0730 100644 --- a/apisix/discovery/kubernetes/core.lua +++ b/apisix/discovery/kubernetes/core.lua @@ -236,16 +236,12 @@ end -- ─── endpoint dict operations (exported) ────────────────────────────── function _M.update_endpoint_dict(handle, endpoints, endpoint_key) + -- one key per service, written by a single atomic safe_set: workers + -- re-read it when their cache entry expires, so there is no version key + -- that could be observed before the content (apache/apisix#14149) local endpoint_content = core.json.encode(endpoints, true) - local endpoint_version = ngx.crc32_long(endpoint_content) - local _, err - _, err = handle.endpoint_dict:safe_set(endpoint_key .. "#version", endpoint_version) + local _, err = handle.endpoint_dict:safe_set(endpoint_key, endpoint_content) if err then - return false, "set endpoint version into discovery DICT failed, " .. err - end - _, err = handle.endpoint_dict:safe_set(endpoint_key, endpoint_content) - if err then - handle.endpoint_dict:delete(endpoint_key .. "#version") return false, "set endpoint into discovery DICT failed, " .. err end return true @@ -255,8 +251,7 @@ end function _M.create_endpoint_lrucache(endpoint_dict, endpoint_key, endpoint_port) local endpoint_content = endpoint_dict:get(endpoint_key) if not endpoint_content then - core.log.error("get empty endpoint content from discovery DIC, this should not happen ", - endpoint_key) + core.log.info("endpoint not found in discovery DICT, key: ", endpoint_key) return nil end @@ -409,7 +404,6 @@ function _M.create_endpoint_callbacks(options) end if operate == "list" then handle.current_keys_hash[endpoint_key] = true - handle.current_keys_hash[endpoint_key .. "#version"] = true end end @@ -508,7 +502,6 @@ function _M.create_endpoint_callbacks(options) end if operate == "list" then handle.current_keys_hash[endpoint_key] = true - handle.current_keys_hash[endpoint_key .. "#version"] = true end end @@ -528,7 +521,6 @@ function _M.create_endpoint_callbacks(options) core.log.debug(core.json.delay_encode(endpoint)) local endpoint_key = build_endpoint_key( key_prefix, endpoint.metadata.namespace, endpoint.metadata.name) - handle.endpoint_dict:delete(endpoint_key .. "#version") handle.endpoint_dict:delete(endpoint_key) end @@ -711,13 +703,9 @@ function _M.resolve_nodes(endpoint_lrucache, service_name, pattern, dict_resolve return nil end - local endpoint_version = endpoint_dict:get(endpoint_key .. "#version") - if not endpoint_version then - core.log.info("get empty endpoint version from discovery DICT ", endpoint_key) - return nil - end - - return endpoint_lrucache(service_name, endpoint_version, + -- no version probe: the entry is re-read from the shared dict when its + -- ttl expires (see the lrucache options in init.lua) + return endpoint_lrucache(service_name, nil, _M.create_endpoint_lrucache, endpoint_dict, endpoint_key, endpoint_port) end diff --git a/apisix/discovery/kubernetes/init.lua b/apisix/discovery/kubernetes/init.lua index 11c8b0630..ffbcce2ab 100644 --- a/apisix/discovery/kubernetes/init.lua +++ b/apisix/discovery/kubernetes/init.lua @@ -31,9 +31,19 @@ local k8s_core = require("apisix.discovery.kubernetes.core") local ctx +-- The shared dict is the source of truth. A worker serves a cached node list +-- for at most ENDPOINT_CACHE_TTL seconds and then re-reads it from the dict +-- (invalid_stale: an expired entry is dropped, never handed back), so there +-- is no version probe and no way to cache a torn pair (apache/apisix#14149). +-- A miss is cached for the same time so an unknown service does not hit the +-- dict on every request. +local ENDPOINT_CACHE_TTL = 1 local endpoint_lrucache = core.lrucache.new({ - ttl = 300, - count = 1024 + ttl = ENDPOINT_CACHE_TTL, + count = 1024, + invalid_stale = true, + neg_ttl = ENDPOINT_CACHE_TTL, + neg_count = 64, }) @@ -164,9 +174,13 @@ end local function merge_cluster_nodes(endpoint_dicts, endpoint_key, endpoint_port) local nodes = {} local seen = {} + local found = false for _, endpoint_dict in ipairs(endpoint_dicts) do local cluster_nodes = k8s_core.create_endpoint_lrucache(endpoint_dict, endpoint_key, endpoint_port) + if cluster_nodes then + found = true + end for _, node in ipairs(cluster_nodes or {}) do local addr = node.host .. ":" .. tostring(node.port) if not seen[addr] then @@ -176,6 +190,12 @@ local function merge_cluster_nodes(endpoint_dicts, endpoint_key, endpoint_port) end end + -- nil, not an empty table, when no selected cluster knows the service, + -- so that the miss is negative-cached like in single cluster mode + if not found then + return nil + end + return nodes end @@ -203,7 +223,6 @@ local function selected_clusters_nodes(service_name, cluster_ids) local endpoint_key, endpoint_port = match[1], match[2] local endpoint_dicts = core.table.new(#cluster_ids, 0) - local versions = core.table.new(#cluster_ids, 0) local unknown_ids for _, id in ipairs(cluster_ids) do local endpoint_dict = ctx[id] @@ -211,11 +230,7 @@ local function selected_clusters_nodes(service_name, cluster_ids) unknown_ids = unknown_ids or {} core.table.insert(unknown_ids, id) else - local endpoint_version = endpoint_dict:get(endpoint_key .. "#version") - if endpoint_version then - core.table.insert(endpoint_dicts, endpoint_dict) - core.table.insert(versions, id .. "#" .. endpoint_version) - end + core.table.insert(endpoint_dicts, endpoint_dict) end end @@ -225,12 +240,12 @@ local function selected_clusters_nodes(service_name, cluster_ids) end if #endpoint_dicts == 0 then - core.log.info("get empty endpoint version from selected clusters for ", service_name) + core.log.info("no known kubernetes discovery cluster selected for ", service_name) return nil end - return endpoint_lrucache(service_name .. "#" .. core.table.concat(cluster_ids, ","), - core.table.concat(versions, ","), + -- merged from the selected dicts, refreshed when the cache entry expires + return endpoint_lrucache(service_name .. "#" .. core.table.concat(cluster_ids, ","), nil, merge_cluster_nodes, endpoint_dicts, endpoint_key, endpoint_port) end diff --git a/apisix/discovery/tars/init.lua b/apisix/discovery/tars/init.lua index 17bb2758b..e5cacd8e5 100644 --- a/apisix/discovery/tars/init.lua +++ b/apisix/discovery/tars/init.lua @@ -55,9 +55,15 @@ local default_weight local last_fetch_full_time = 0 local last_db_error +-- same model as the kubernetes discovery: the shared dict is the source of +-- truth and a cached entry is re-read when it expires, no version probe +local ENDPOINT_CACHE_TTL = 1 local endpoint_lrucache = core.lrucache.new({ - ttl = 300, - count = 1024 + ttl = ENDPOINT_CACHE_TTL, + count = 1024, + invalid_stale = true, + neg_ttl = ENDPOINT_CACHE_TTL, + neg_count = 64, }) local activated_buffer = core.table.new(10, 0) @@ -77,25 +83,16 @@ local endpoints_pattern = core.table.concat( local function update_endpoint(servant, nodes) local endpoint_content = core.json.encode(nodes, true) - local endpoint_version = ngx.crc32_long(endpoint_content) core.log.debug("set servant ", servant, endpoint_content) - local _, err - _, err = endpoint_dict:safe_set(servant .. "#version", endpoint_version) - if err then - core.log.error("set endpoint version into nginx shared dict failed, ", err) - return - end - _, err = endpoint_dict:safe_set(servant, endpoint_content) + local _, err = endpoint_dict:safe_set(servant, endpoint_content) if err then core.log.error("set endpoint into nginx shared dict failed, ", err) - endpoint_dict:delete(servant .. "#version") end end local function delete_endpoint(servant) core.log.info("delete servant ", servant) - endpoint_dict:delete(servant .. "#version") endpoint_dict:delete(servant) end @@ -103,7 +100,7 @@ end local function add_endpoint_to_lrucache(servant) local endpoint_content, err = endpoint_dict:get_stale(servant) if not endpoint_content then - core.log.error("get empty endpoint content, servant: ", servant, ", err: ", err) + core.log.info("endpoint not found in tars DICT, servant: ", servant, ", err: ", err) return nil end @@ -129,14 +126,7 @@ local function get_endpoint(servant) so we must use endpoint_dict:get_stale() to get value instead endpoint_dict:get() --]] - local endpoint_version, err = endpoint_dict:get_stale(servant .. "#version") - if not endpoint_version then - if err then - core.log.error("get empty endpoint version, servant: ", servant, ", err: ", err) - end - return nil - end - return endpoint_lrucache(servant, endpoint_version, add_endpoint_to_lrucache, servant) + return endpoint_lrucache(servant, nil, add_endpoint_to_lrucache, servant) end diff --git a/docs/en/latest/discovery/kubernetes.md b/docs/en/latest/discovery/kubernetes.md index b00e3750f..92589dbb1 100644 --- a/docs/en/latest/discovery/kubernetes.md +++ b/docs/en/latest/discovery/kubernetes.md @@ -327,7 +327,7 @@ Because lua-resty-http does not currently support mTLS, and basic authentication **Q: APISIX inherits Nginx's multiple process model, does it mean that each nginx worker process will [_List-Watch_](https://kubernetes.io/docs/reference/using-api/api-concepts/) kubernetes endpoints resources?** -A: The Kubernetes service discovery only uses privileged processes to [_List-Watch_](https://kubernetes.io/docs/reference/using-api/api-concepts/) Kubernetes endpoints resources, then store theirs value into `ngx.shared.DICT`, worker processes get results by querying `ngx.shared.DICT`. +A: The Kubernetes service discovery only uses privileged processes to [_List-Watch_](https://kubernetes.io/docs/reference/using-api/api-concepts/) Kubernetes endpoints resources, then store theirs value into `ngx.shared.DICT`, worker processes get results by querying `ngx.shared.DICT`. Each worker process caches the nodes of a service for one second and then reads them from `ngx.shared.DICT` again, so a change reaches every worker process within one second of the privileged process writi [...] **Q: What permissions do [_ServiceAccount_](https://kubernetes.io/docs/tasks/configure-pod-container/configure-service-account/) require?** diff --git a/docs/zh/latest/discovery/kubernetes.md b/docs/zh/latest/discovery/kubernetes.md index 6ec9993ed..beae487db 100644 --- a/docs/zh/latest/discovery/kubernetes.md +++ b/docs/zh/latest/discovery/kubernetes.md @@ -325,7 +325,7 @@ A: 一般情况下，我们有三种方式可以完成与 Kubernetes APIServer **Q: APISIX 继承了 NGINX 的多进程模型，是否意味着每个 APISIX 工作进程都会监听 Kubernetes Endpoints？** -A: Kubernetes 服务发现只使用特权进程监听 Kubernetes Endpoints，然后将其值存储到 `ngx.shared.DICT` 中，工作进程通过查询 `ngx.shared.DICT` 来获取结果。 +A: Kubernetes 服务发现只使用特权进程监听 Kubernetes Endpoints，然后将其值存储到 `ngx.shared.DICT` 中，工作进程通过查询 `ngx.shared.DICT` 来获取结果。每个工作进程会把一个服务的节点缓存 1 秒，到期后重新从 `ngx.shared.DICT` 读取，因此一次变化会在特权进程写入 `ngx.shared.DICT` 后 1 秒内到达所有工作进程。 **Q: [_ServiceAccount_](https://kubernetes.io/docs/tasks/configure-pod-container/configure-service-account/) 需要的权限有哪些？** diff --git a/t/kubernetes/discovery/kubernetes2.t b/t/kubernetes/discovery/kubernetes2.t index 41a3819d4..8bdd024bd 100644 --- a/t/kubernetes/discovery/kubernetes2.t +++ b/t/kubernetes/discovery/kubernetes2.t @@ -105,7 +105,7 @@ _EOC_ local core = require("apisix.core") local d = require("apisix.discovery.kubernetes") - ngx.sleep(1) + ngx.sleep(1.1) ngx.req.read_body() local request_body = ngx.req.get_body_data() diff --git a/t/kubernetes/discovery/kubernetes3.t b/t/kubernetes/discovery/kubernetes3.t index 47b902e18..d7972e08c 100644 --- a/t/kubernetes/discovery/kubernetes3.t +++ b/t/kubernetes/discovery/kubernetes3.t @@ -127,7 +127,7 @@ _EOC_ local core = require("apisix.core") local d = require("apisix.discovery.kubernetes") - ngx.sleep(1) + ngx.sleep(1.1) ngx.req.read_body() local request_body = ngx.req.get_body_data() @@ -205,7 +205,7 @@ _EOC_ return 500 end - ngx.sleep(1) + ngx.sleep(1.1) local k8s = require("apisix.discovery.kubernetes") local data = k8s.dump_data() @@ -226,7 +226,7 @@ _EOC_ local http = require "resty.http" local httpc = http.new() - ngx.sleep(1) + ngx.sleep(1.1) local dump_uri = "http://127.0.0.1:"; .. ngx.var.server_port .. "/v1/discovery/kubernetes/dump" local res, err = httpc:request_uri(dump_uri, { method = "GET"}) @@ -253,7 +253,7 @@ _EOC_ ngx.status = res.status return end - ngx.sleep(1) + ngx.sleep(1.1) end local httpc = http.new() local res, _ = httpc:request_uri(healthcheck_uri, {method = "GET", keepalive = false}) diff --git a/t/kubernetes/discovery/kubernetes4.t b/t/kubernetes/discovery/kubernetes4.t index 3214523ac..266f4992d 100644 --- a/t/kubernetes/discovery/kubernetes4.t +++ b/t/kubernetes/discovery/kubernetes4.t @@ -255,7 +255,7 @@ _EOC_ local core = require("apisix.core") local d = require("apisix.discovery.kubernetes") - ngx.sleep(1) + ngx.sleep(1.1) ngx.req.read_body() local request_body = ngx.req.get_body_data() @@ -337,7 +337,7 @@ _EOC_ return 500 end - ngx.sleep(1) + ngx.sleep(1.1) if res.status ~= 200 and res.status ~= 201 and res.status ~= 409 then return res.status diff --git a/t/kubernetes/discovery/kubernetes5.t b/t/kubernetes/discovery/kubernetes5.t index 572494343..555b99641 100644 --- a/t/kubernetes/discovery/kubernetes5.t +++ b/t/kubernetes/discovery/kubernetes5.t @@ -109,7 +109,7 @@ _EOC_ local core = require("apisix.core") local d = require("apisix.discovery.kubernetes") - ngx.sleep(1) + ngx.sleep(1.1) ngx.req.read_body() local request_body = ngx.req.get_body_data() @@ -285,10 +285,9 @@ skip unknown kubernetes discovery cluster ids: third, service: ns-b/ep:p1 location /t { content_by_lua_block { local core = require("apisix.core") - local function set_endpoints(id, key, endpoints, version) + local function set_endpoints(id, key, endpoints) local dict = ngx.shared["kubernetes-" .. id] dict:set(key, core.json.encode(endpoints)) - dict:set(key .. "#version", version) end local d = require("apisix.discovery.kubernetes") @@ -309,14 +308,14 @@ skip unknown kubernetes discovery cluster ids: third, service: ns-b/ep:p1 set_endpoints("first", "ns/svc", {p1 = { {host = "10.0.0.1", port = 80, weight = 50}, {host = "10.0.0.9", port = 80, weight = 50}, - }}, "1") + }}) set_endpoints("second", "ns/svc", {p1 = { {host = "10.0.0.2", port = 80, weight = 50}, {host = "10.0.0.9", port = 80, weight = 50}, - }}, "1") + }}) set_endpoints("first", "ns/only-first", {p1 = { {host = "10.0.1.1", port = 80, weight = 50}, - }}, "1") + }}) show("ns/svc:p1", {"first", "second"}) show("ns/svc:p1", {"first"}) @@ -328,7 +327,9 @@ skip unknown kubernetes discovery cluster ids: third, service: ns-b/ep:p1 set_endpoints("second", "ns/svc", {p1 = { {host = "10.0.0.2", port = 80, weight = 50}, {host = "10.0.0.3", port = 80, weight = 50}, - }}, "2") + }}) + -- a worker re-reads the shared dicts once its cache entry expires + ngx.sleep(1.1) show("ns/svc:p1", {"first", "second"}) } } @@ -386,22 +387,21 @@ routes: location /t { content_by_lua_block { local core = require("apisix.core") - local function set_endpoints(id, key, endpoints, version) + local function set_endpoints(id, key, endpoints) local dict = ngx.shared["kubernetes-" .. id] dict:set(key, core.json.encode(endpoints)) - dict:set(key .. "#version", version) end local http = require("resty.http") set_endpoints("first", "ns/svc", {p1 = { {host = "127.0.0.1", port = 1979, weight = 50}, - }}, "1") + }}) set_endpoints("second", "ns/svc", {p1 = { {host = "127.0.0.1", port = 1980, weight = 50}, - }}, "1") + }}) set_endpoints("second", "ns/only-second", {p1 = { {host = "127.0.0.1", port = 1980, weight = 50}, - }}, "1") + }}) local uri = "http://127.0.0.1:"; .. ngx.var.server_port for _, path in ipairs({"/hello", "/hello", "/hello", "/hello1", "/hello_chunked"}) do @@ -454,7 +454,6 @@ discovery: dict:set("ns/svc", core.json.encode({p1 = { {host = "10.0.0.1", port = 80, weight = 50}, }})) - dict:set("ns/svc#version", "1") local nodes = d.nodes("ns/svc:p1") ngx.say(#nodes) @@ -511,7 +510,6 @@ routes: dict:set("ns/svc", core.json.encode({p1 = { {host = "127.0.0.1", port = 1980, weight = 50}, }})) - dict:set("ns/svc#version", "1") local uri = "http://127.0.0.1:"; .. ngx.var.server_port for _, path in ipairs({"/hello", "/hello1"}) do @@ -556,20 +554,19 @@ stream_routes: local core = require("apisix.core") local d = require("apisix.discovery.kubernetes") - local function set_endpoints(id, key, endpoints, version) + local function set_endpoints(id, key, endpoints) local dict = ngx.shared["kubernetes-" .. id .. "-stream"] dict:set(key, core.json.encode(endpoints)) - dict:set(key .. "#version", version) end set_endpoints("first", "ns/svc", {p1 = { {host = "127.0.0.1", port = 1979, weight = 50}, {host = "127.0.0.1", port = 1995, weight = 50}, - }}, "1") + }}) set_endpoints("second", "ns/svc", {p1 = { {host = "127.0.0.1", port = 1995, weight = 50}, {host = "127.0.0.2", port = 1995, weight = 50}, - }}, "1") + }}) local nodes = d.nodes("ns/svc:p1", {cluster_ids = {"first", "second"}}) local addrs = {} diff --git a/t/kubernetes/discovery/kubernetes6.t b/t/kubernetes/discovery/kubernetes6.t new file mode 100644 index 000000000..750362694 --- /dev/null +++ b/t/kubernetes/discovery/kubernetes6.t @@ -0,0 +1,372 @@ +# +# 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. +# + +# End-to-end checks for the kubernetes discovery cache (apache/apisix#14149): +# the Endpoints object is changed through the real apiserver (kubectl proxy +# on 6445) and real requests are sent through APISIX to the fake backend on +# 1980/1981, so every worker's cache is exercised, not just the shared dict. +# +# Kubernetes rejects loopback addresses in Endpoints, so the fake backend is +# addressed through the first global IPv4 address of this host; the backend +# listens on every interface. + +BEGIN { + # listen with reuseport so that the probe connections are spread over + # the four workers instead of landing on whichever worker wakes first + $ENV{TEST_NGINX_REUSE_PORT} = 1; + + our $token_file = "/tmp/var/run/secrets/kubernetes.io/serviceaccount/token"; + our $token_value = eval {`cat $token_file 2>/dev/null`}; + our $backend_ip = eval {`ip -4 -o addr show scope global 2>/dev/null | awk '{print \$4}' | cut -d/ -f1 | head -n 1`}; + chomp($backend_ip) if defined $backend_ip; + + our $yaml_config = <<_EOC_; +apisix: + node_listen: 1984 +deployment: + role: data_plane + role_data_plane: + config_provider: yaml +discovery: + kubernetes: + service: + host: "127.0.0.1" + port: "6443" + ssl_verify: false + client: + token_file: "/tmp/var/run/secrets/kubernetes.io/serviceaccount/token" +_EOC_ +} + +use t::APISIX 'no_plan'; + +repeat_each(1); +log_level('warn'); +no_root_location(); +no_shuffle(); +workers(4); + +add_block_preprocessor(sub { + my ($block) = @_; + + my $apisix_yaml = $block->apisix_yaml // <<_EOC_; +routes: + - + uri: /server_port + upstream: + service_name: ns-a/ep:p1 + discovery_type: kubernetes + type: roundrobin + timeout: + connect: 1 + send: 1 + read: 1 + - + uri: /hello + upstream: + service_name: ns-a/no-such-ep:p1 + discovery_type: kubernetes + type: roundrobin + timeout: + connect: 1 + send: 1 + read: 1 +#END +_EOC_ + + $block->set_value("apisix_yaml", $apisix_yaml); + + my $main_config = $block->main_config // <<_EOC_; +env KUBERNETES_SERVICE_HOST=127.0.0.1; +env KUBERNETES_SERVICE_PORT=6443; +env KUBERNETES_CLIENT_TOKEN=$::token_value; +env KUBERNETES_CLIENT_TOKEN_FILE=$::token_file; +env E2E_BACKEND_IP=$::backend_ip; +_EOC_ + + $block->set_value("main_config", $main_config); + + if (!$block->request) { + $block->set_value("request", "GET /t"); + } + + if (!$block->timeout) { + $block->set_value("timeout", "60"); + } +}); + +run_tests(); + +__DATA__ + +=== TEST 1: a Service published through the apiserver is served by every worker +# The first probes may run before the initial list has finished or hit the +# addresses a previous test file left behind, which logs an [error] for a +# 503, so the default `no_error_log: [error]` would make this block flaky. +--- yaml_config eval: $::yaml_config +--- config + location /t { + content_by_lua_block { + local t = require("lib.k8s_discovery_e2e").new(ngx.var.server_port) + + ngx.say("backend ip known: ", t.backend_ip() ~= nil) + local status = t.ensure_endpoints(t.port_subsets(1980)) + ngx.say("published: ", status == 200 or status == 201) + ngx.say("serving 1980 within 5s: ", t.wait_for("1980", 5, 5) ~= nil) + ngx.say("answers not from 1980 in 40 requests: ", t.count_other_than("1980", 40)) + } + } +--- response_body +backend ip known: true +published: true +serving 1980 within 5s: true +answers not from 1980 in 40 requests: 0 +--- no_error_log +[alert] + + + +=== TEST 2: a change of the Endpoints reaches every worker within the cache ttl +# The first probes may run before the initial list has finished or hit the +# addresses a previous test file left behind, which logs an [error] for a +# 503, so the default `no_error_log: [error]` would make this block flaky. +--- yaml_config eval: $::yaml_config +--- config + location /t { + content_by_lua_block { + local t = require("lib.k8s_discovery_e2e").new(ngx.var.server_port) + + t.ensure_endpoints(t.port_subsets(1980)) + ngx.say("serving 1980 within 5s: ", t.wait_for("1980", 5, 5) ~= nil) + + ngx.say("patch: ", t.set_subsets(t.port_subsets(1981))) + local elapsed, unexpected = t.wait_for("1981", 5, 5, {["1980"] = true, ["1981"] = true}) + ngx.say("switched to 1981 within 5s: ", elapsed ~= nil) + ngx.say("answers that were neither 1980 nor 1981 during the switch: ", unexpected) + ngx.say("answers not from 1981 in 40 requests: ", t.count_other_than("1981", 40)) + } + } +--- response_body +serving 1980 within 5s: true +patch: 200 +switched to 1981 within 5s: true +answers that were neither 1980 nor 1981 during the switch: 0 +answers not from 1981 in 40 requests: 0 +--- no_error_log +[alert] + + + +=== TEST 3: scaling the Endpoints to zero and back +--- yaml_config eval: $::yaml_config +--- config + location /t { + content_by_lua_block { + local t = require("lib.k8s_discovery_e2e").new(ngx.var.server_port) + + t.ensure_endpoints(t.port_subsets(1980)) + ngx.say("serving 1980 within 5s: ", t.wait_for("1980", 5, 5) ~= nil) + + ngx.say("patch to empty: ", t.set_subsets({})) + ngx.say("503 within 5s: ", t.wait_for("503", 5, 5) ~= nil) + + ngx.say("patch back: ", t.set_subsets(t.port_subsets(1980))) + ngx.say("serving 1980 again within 5s: ", t.wait_for("1980", 5, 5) ~= nil) + } + } +--- response_body +serving 1980 within 5s: true +patch to empty: 200 +503 within 5s: true +patch back: 200 +serving 1980 again within 5s: true +--- error_log +failed to set upstream + + + +=== TEST 4: deleting and recreating the Endpoints object +--- yaml_config eval: $::yaml_config +--- config + location /t { + content_by_lua_block { + local t = require("lib.k8s_discovery_e2e").new(ngx.var.server_port) + + t.ensure_endpoints(t.port_subsets(1980)) + ngx.say("serving 1980 within 5s: ", t.wait_for("1980", 5, 5) ~= nil) + + ngx.say("delete: ", t.delete_endpoints()) + ngx.say("503 within 5s: ", t.wait_for("503", 5, 5) ~= nil) + + ngx.say("create: ", t.create_endpoints(t.port_subsets(1980))) + ngx.say("serving 1980 again within 5s: ", t.wait_for("1980", 5, 5) ~= nil) + } + } +--- response_body +serving 1980 within 5s: true +delete: 200 +503 within 5s: true +create: 201 +serving 1980 again within 5s: true +--- error_log +failed to set upstream + + + +=== TEST 5: unknown service, dump without version keys, fixture restored +--- yaml_config eval: $::yaml_config +--- config + location /t { + content_by_lua_block { + local t = require("lib.k8s_discovery_e2e").new(ngx.var.server_port) + + t.ensure_endpoints(t.port_subsets(1980)) + ngx.say("serving 1980 within 5s: ", t.wait_for("1980", 5, 5) ~= nil) + + ngx.say("unknown service: ", t.get("/hello")) + + local names = t.dump_names() + ngx.say("dump has ns-a/ep: ", names["ns-a/ep"] == true) + + -- the writer is the real privileged agent here, so look at the + -- shared dict itself, not at the dump, which filters version keys + local version_keys = 0 + for _, key in ipairs(ngx.shared["kubernetes"]:get_keys(0)) do + if key:sub(-8) == "#version" then + version_keys = version_keys + 1 + end + end + ngx.say("version keys in the shared dict: ", version_keys) + + local status = t.ensure_endpoints(t.fixture_subsets()) + ngx.say("fixture restored: ", status == 200 or status == 201) + } + } +--- response_body +serving 1980 within 5s: true +unknown service: 503 +dump has ns-a/ep: true +version keys in the shared dict: 0 +fixture restored: true +--- error_log +failed to set upstream + + + +=== TEST 6: every worker resolves the new nodes after a change +# The issue was diagnosed per worker with ngx.worker.id(): one worker kept +# a stale list while the others were fine. /worker_nodes answers from the +# worker that served the request, and the probes are spread over the four +# workers by reuseport, so each worker is checked on its own. The first +# probes may run before the initial list has finished, which logs an +# [error] for a 503, so the default `no_error_log: [error]` would be flaky. +--- yaml_config eval: $::yaml_config +--- config + location /worker_nodes { + content_by_lua_block { + local d = require("apisix.discovery.kubernetes") + local addrs = {} + for _, node in ipairs(d.nodes("ns-a/ep:p1") or {}) do + table.insert(addrs, node.host .. ":" .. node.port) + end + ngx.print(ngx.worker.id(), " ", table.concat(addrs, ",")) + } + } + + location /t { + content_by_lua_block { + local t = require("lib.k8s_discovery_e2e").new(ngx.var.server_port) + local ip = t.backend_ip() + + t.ensure_endpoints(t.port_subsets(1980)) + ngx.say("serving 1980 within 5s: ", t.wait_for("1980", 5, 5) ~= nil) + local ok, seen = t.wait_for_workers(ip .. ":1980", 4, 10) + ngx.say("every worker on 1980: ", ok, ", workers seen: ", seen) + + ngx.say("patch: ", t.set_subsets(t.port_subsets(1981))) + ok, seen = t.wait_for_workers(ip .. ":1981", 4, 10) + ngx.say("every worker on 1981: ", ok, ", workers seen: ", seen) + } + } +--- response_body +serving 1980 within 5s: true +every worker on 1980: true, workers seen: 4 +patch: 200 +every worker on 1981: true, workers seen: 4 +--- no_error_log +[alert] + + + +=== TEST 7: a burst of changes under traffic leaves no worker behind +# The reported incidents followed rollouts and bulk scale-downs: many +# Endpoints changes in a short time while requests keep flowing. The same +# [error] caveat as above applies to the first probes. +--- yaml_config eval: $::yaml_config +--- config + location /worker_nodes { + content_by_lua_block { + local d = require("apisix.discovery.kubernetes") + local addrs = {} + for _, node in ipairs(d.nodes("ns-a/ep:p1") or {}) do + table.insert(addrs, node.host .. ":" .. node.port) + end + ngx.print(ngx.worker.id(), " ", table.concat(addrs, ",")) + } + } + + location /t { + content_by_lua_block { + local t = require("lib.k8s_discovery_e2e").new(ngx.var.server_port) + local ip = t.backend_ip() + + t.ensure_endpoints(t.port_subsets(1980)) + ngx.say("serving 1980 within 5s: ", t.wait_for("1980", 5, 5) ~= nil) + + -- twelve changes back and forth, three requests after each one; + -- the burst ends on 1980 + local allowed = {["1980"] = true, ["1981"] = true} + local patches, unexpected = 0, 0 + for i = 1, 12 do + local port = (i % 2 == 1) and 1981 or 1980 + if t.set_subsets(t.port_subsets(port)) == 200 then + patches = patches + 1 + end + for _ = 1, 3 do + if not allowed[t.get("/server_port")] then + unexpected = unexpected + 1 + end + end + end + ngx.say("patches applied: ", patches) + ngx.say("answers outside 1980 and 1981 during the burst: ", unexpected) + + -- let the last event land and every cache entry expire + ngx.sleep(1.5) + local ok, seen = t.wait_for_workers(ip .. ":1980", 4, 10) + ngx.say("every worker on the final state: ", ok, ", workers seen: ", seen) + ngx.say("answers not from 1980 in 40 requests: ", t.count_other_than("1980", 40)) + } + } +--- response_body +serving 1980 within 5s: true +patches applied: 12 +answers outside 1980 and 1981 during the burst: 0 +every worker on the final state: true, workers seen: 4 +answers not from 1980 in 40 requests: 0 +--- no_error_log +[alert] diff --git a/t/kubernetes/discovery/kubernetes7.t b/t/kubernetes/discovery/kubernetes7.t new file mode 100644 index 000000000..713ee6437 --- /dev/null +++ b/t/kubernetes/discovery/kubernetes7.t @@ -0,0 +1,135 @@ +# +# 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. +# + +# no apiserver listens on 6446, so the tests below own the endpoint dict +BEGIN { + our $offline_yaml_config = <<_EOC_; +apisix: + node_listen: 1984 +deployment: + role: data_plane + role_data_plane: + config_provider: yaml +discovery: + kubernetes: + service: + schema: "http" + host: "127.0.0.1" + port: "6446" + client: + token: "fake" +_EOC_ +} + +use t::APISIX 'no_plan'; + +repeat_each(1); +log_level('warn'); +no_root_location(); +no_shuffle(); + +add_block_preprocessor(sub { + my ($block) = @_; + + my $apisix_yaml = $block->apisix_yaml // <<_EOC_; +routes: [] +#END +_EOC_ + + $block->set_value("apisix_yaml", $apisix_yaml); + + if (!$block->request) { + $block->set_value("request", "GET /t"); + } +}); + +run_tests(); + +__DATA__ + +=== TEST 1: a worker re-reads the shared dict once its cache entry expires (#14149) +# The dict holds one key per service and is the source of truth; there is no +# version key a worker could observe before the content. +--- yaml_config eval: $::offline_yaml_config +--- config + location /t { + content_by_lua_block { + local core = require("apisix.core") + local d = require("apisix.discovery.kubernetes") + local dict = ngx.shared["kubernetes"] + + local function publish(ip) + dict:set("ns/svc", core.json.encode({p1 = { + {host = ip, port = 80, weight = 50}, + }})) + end + local function first_host() + local nodes = d.nodes("ns/svc:p1") + return nodes and nodes[1].host or "nil" + end + + publish("10.0.0.1") + ngx.say(first_host()) + publish("10.0.0.2") + ngx.say(first_host()) + ngx.sleep(1.2) + ngx.say(first_host()) + } + } +--- response_body +10.0.0.1 +10.0.0.1 +10.0.0.2 +--- error_log +connect apiserver failed + + + +=== TEST 2: the writer publishes exactly one key per service +--- yaml_config eval: $::offline_yaml_config +--- config + location /t { + content_by_lua_block { + local k8s_core = require("apisix.discovery.kubernetes.core") + local real_dict = ngx.shared["kubernetes"] + + -- record every write that reaches the shared dict + local writing = {set = true, safe_set = true, add = true, + safe_add = true, replace = true} + local writes = {} + local recording_dict = setmetatable({}, { + __index = function(_, method) + return function(_, key, ...) + if writing[method] then + table.insert(writes, method .. " " .. key) + end + return real_dict[method](real_dict, key, ...) + end + end, + }) + + local ok, err = k8s_core.update_endpoint_dict( + {endpoint_dict = recording_dict}, + {p1 = {{host = "10.0.0.1", port = 80, weight = 50}}}, + "ns/svc2") + ngx.say(ok, " ", err or "", " ", table.concat(writes, ",")) + } + } +--- response_body +true safe_set ns/svc2 +--- error_log +connect apiserver failed diff --git a/t/kubernetes/discovery/stream/kubernetes.t b/t/kubernetes/discovery/stream/kubernetes.t index a9058f55d..e8a9086f4 100644 --- a/t/kubernetes/discovery/stream/kubernetes.t +++ b/t/kubernetes/discovery/stream/kubernetes.t @@ -157,7 +157,7 @@ _EOC_ local core = require("apisix.core") local d = require("apisix.discovery.kubernetes") - ngx.sleep(1) + ngx.sleep(1.1) local sock = ngx.req.socket() local request_body = sock:receive() diff --git a/t/lib/k8s_discovery_e2e.lua b/t/lib/k8s_discovery_e2e.lua new file mode 100644 index 000000000..e49b93603 --- /dev/null +++ b/t/lib/k8s_discovery_e2e.lua @@ -0,0 +1,235 @@ +-- +-- 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. +-- + +-- helpers for t/kubernetes/discovery/kubernetes6.t: change the ns-a/ep +-- Endpoints through the kubectl proxy and send real requests through APISIX + +local core = require("apisix.core") +local http = require("resty.http") +local ngx = ngx +local ipairs = ipairs +local pairs = pairs +local tostring = tostring +local os_getenv = os.getenv + +local _M = {} + +local apiserver = "http://127.0.0.1:6445"; +local ep_path = "/api/v1/namespaces/ns-a/endpoints/ep" +local ep_collection = "/api/v1/namespaces/ns-a/endpoints" + + +function _M.new(gateway_port) + local t = {} + local gateway = "http://127.0.0.1:"; .. tostring(gateway_port) + + local function k8s(method, path, body, content_type) + local httpc = http.new() + httpc:set_timeouts(5000, 5000, 5000) + local res, err = httpc:request_uri(apiserver .. path, { + method = method, + body = body, + headers = {["Content-Type"] = content_type or "application/json"}, + }) + if not res then + return "err:" .. tostring(err) + end + return res.status + end + + -- Kubernetes rejects loopback addresses in Endpoints, so the fake + -- backend is reached through a global address of this host, exported + -- by the test file as E2E_BACKEND_IP + function t.backend_ip() + local ip = os_getenv("E2E_BACKEND_IP") + if ip == nil or ip == "" then + return nil + end + return ip + end + + function t.port_subsets(port) + return {{ + addresses = {{ip = t.backend_ip() or "127.0.0.1"}}, + ports = {{name = "p1", port = port}}, + }} + end + + -- the subsets kubernetes2.t installs, so later test files find them + function t.fixture_subsets() + return { + { + addresses = {{ip = "10.0.0.1"}, {ip = "10.0.0.2"}}, + ports = {{name = "p1", port = 5001}}, + }, + { + addresses = {{ip = "20.0.0.1"}, {ip = "20.0.0.2"}}, + ports = {{name = "p2", port = 5002}}, + }, + } + end + + function t.set_subsets(subsets) + local patch + if #subsets == 0 then + patch = '[{"op":"replace","path":"/subsets","value":[]}]' + else + patch = core.json.encode({{op = "replace", path = "/subsets", value = subsets}}, true) + end + return k8s("PATCH", ep_path, patch, "application/json-patch+json") + end + + function t.delete_endpoints() + return k8s("DELETE", ep_path) + end + + -- patch the subsets, or create the Endpoints object when it is gone + -- (a previous run may have deleted it); 200 or 201 on success + function t.ensure_endpoints(subsets) + local status = t.set_subsets(subsets) + if status == 404 then + return t.create_endpoints(subsets) + end + return status + end + + function t.create_endpoints(subsets) + local body = core.json.encode({ + apiVersion = "v1", + kind = "Endpoints", + metadata = {name = "ep", namespace = "ns-a"}, + subsets = subsets, + }, true) + return k8s("POST", ep_collection, body) + end + + -- one request on a fresh connection, so that different requests can be + -- served by different workers; returns the body on 200, else the status + function t.get(path) + local httpc = http.new() + -- a probe proxied to a stale, unroutable address must fail fast + httpc:set_timeouts(2000, 2000, 2000) + local res, err = httpc:request_uri(gateway .. path, {keepalive = false}) + if not res then + return "err:" .. tostring(err) + end + if res.status ~= 200 then + return tostring(res.status) + end + return res.body + end + + local function probe() + return t.get("/server_port") + end + + -- polls until n consecutive answers equal expected; returns the seconds + -- it took (nil on timeout) and how many answers were outside allowed + function t.wait_for(expected, n, timeout, allowed) + ngx.update_time() + local start = ngx.now() + local unexpected = 0 + while true do + local ok = true + for _ = 1, n do + local got = probe() + if allowed and not allowed[got] then + unexpected = unexpected + 1 + end + if got ~= expected then + ok = false + break + end + end + ngx.update_time() + local elapsed = ngx.now() - start + if ok then + -- the probes may not have reached every worker yet: give every + -- worker's one second cache entry time to expire before the + -- caller runs its strict checks + ngx.sleep(1.2) + return elapsed, unexpected + end + if elapsed > timeout then + return nil, unexpected + end + ngx.sleep(0.2) + end + end + + -- polls the /worker_nodes location of the test block, which answers + -- "<worker id> <host:port,...>" from the worker that served the request, + -- until every one of n_workers workers has answered and the last answer + -- of each of them equals expected; returns whether that happened and + -- how many distinct workers answered + function t.wait_for_workers(expected, n_workers, timeout) + ngx.update_time() + local start = ngx.now() + local last = {} + while true do + local body = t.get("/worker_nodes") + local id, nodes = body:match("^(%d+) (.*)$") + if id then + last[id] = nodes + end + + local seen, converged = 0, 0 + for _, answer in pairs(last) do + seen = seen + 1 + if answer == expected then + converged = converged + 1 + end + end + if seen >= n_workers and converged == seen then + return true, seen + end + + ngx.update_time() + if ngx.now() - start > timeout then + return false, seen + end + ngx.sleep(0.05) + end + end + + function t.count_other_than(expected, n) + local other = 0 + for _ = 1, n do + if probe() ~= expected then + other = other + 1 + end + end + return other + end + + -- the names in the discovery dump, as a set + function t.dump_names() + local names = {} + local data = require("apisix.discovery.kubernetes").dump_data() + for _, cluster in ipairs(data.endpoints or {}) do + for _, ep in ipairs(cluster.endpoints or {}) do + names[ep.name] = true + end + end + return names + end + + return t +end + + +return _M diff --git a/t/tars/discovery/tars.t b/t/tars/discovery/tars.t index 206e4ccc6..b4a806057 100644 --- a/t/tars/discovery/tars.t +++ b/t/tars/discovery/tars.t @@ -389,3 +389,23 @@ where application = 'A' and server_name = 'AServer' and node_name = '172.16.1.1' [ "{ 1 0 1 1 }\n", ] + + + +=== TEST 12: the shared dict holds one key per servant and no version key +--- timeout: 4 +--- config + location /t { + content_by_lua_block { + ngx.sleep(2) + + -- exactly the servants left by the tests above, nothing else + local keys = ngx.shared["tars"]:get_keys(0) + table.sort(keys) + ngx.say(table.concat(keys, ",")) + } + } +--- request +GET /t +--- response_body +A.AServer.FirstObj,B.BServer.FirstObj,C.CServer.FirstObj
