This is an automated email from the ASF dual-hosted git repository.
shreemaan-abhishek pushed a commit to branch release/3.20
in repository https://gitbox.apache.org/repos/asf/apisix.git
The following commit(s) were added to refs/heads/release/3.20 by this push:
new 4c94c575d fix(discovery): refresh the kubernetes endpoint cache by ttl
instead of a version key (#14167) (#14169)
4c94c575d is described below
commit 4c94c575d1cb3c950287b788636878e10cf3aaae
Author: Shreemaan Abhishek <[email protected]>
AuthorDate: Fri Nov 6 17:52:52 2026 +0800
fix(discovery): refresh the kubernetes endpoint cache by ttl instead of a
version key (#14167) (#14169)
(cherry picked from commit 22ecbc60e4ea780abc0c76ee6396583a26ff7e68)
Co-authored-by: Ming Wen <[email protected]>
---
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
