Skip to content

fix(etcd): do not advance the watch revision on a timeout, and make the recovery reload cheap #13721

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
AlinsRan wants to merge 5 commits into
base: master
Choose a base branch
from
+280 −37
Open

fix(etcd): do not advance the watch revision on a timeout, and make the recovery reload cheap #13721

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
da56b54
fix(etcd): do not advance the watch revision on a timeout
AlinsRan Aug 3, 2026
e7473a7
perf(etcd): reuse unchanged items on a full reload
AlinsRan Aug 3, 2026
a66db76
test(etcd): assert delivery after the timeouts and check etcd setup e…
AlinsRan Aug 4, 2026
99180ea
test(etcd): give the new timeout assertion room and poll for delivery
AlinsRan Aug 4, 2026
c78f3c5
chore(etcd): trim the comments added by this change
AlinsRan Aug 4, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
67 changes: 38 additions & 29 deletions apisix/core/config_etcd.lua
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@ local json = require("apisix.core.json")
local etcd_apisix = require("apisix.core.etcd")
local core_str = require("apisix.core.string")
local new_tab = require("table.new")
local nkeys = require("table.nkeys")
local inspect = require("inspect")
local process = require("ngx.process")
local check_schema = require("apisix.core.schema").check
Expand Down Expand Up @@ -196,18 +197,9 @@ local function do_run_watch(premature)
opts.need_cancel = true
opts.start_revision = watch_ctx.rev

-- get latest revision
local res, err = watch_ctx.cli:readdir(watch_ctx.prefix .. "/phantomkey")
if err then
log.error("failed to get latest revision, err: ", err)
end
local latest_rev
if res and res.body and res.body.header and res.body.header.revision then
latest_rev = tonumber(res.body.header.revision)
else
log.error("failed to get latest revision, res: ", json.delay_encode(res))
end

-- A watch timeout must not advance start_revision: it cannot tell an idle
-- prefix from a stream that died silently, and skipping ahead loses the
-- events etcd already wrote into that stream. See #13067.
log.info("restart watchdir: start_revision=", opts.start_revision)

local res_func, err, http_cli = watch_ctx.cli:watchdir(watch_ctx.prefix, opts)
Expand All @@ -227,12 +219,6 @@ local function do_run_watch(premature)
then
log.error("wait watch event: ", err)
end
if err == "timeout" then
if latest_rev and watch_ctx.rev < latest_rev + 1 then
watch_ctx.rev = latest_rev + 1
log.info("etcd watch timeout, upgrade revision to ", watch_ctx.rev)
end
end
cancel_watch(http_cli)
break
end
Expand Down Expand Up @@ -560,6 +546,7 @@ end
local function load_full_data(self, dir_res, headers, prev_values, prev_values_hash)
local err
local changed = false
local prev_keys_still_present = 0

if self.single_item then
self.values = new_tab(1, 0)
Expand Down Expand Up @@ -621,6 +608,22 @@ local function load_full_data(self, dir_res, headers, prev_values, prev_values_h

for _, item in ipairs(values) do
local key = short_key(self, item.key)
local prev_item = get_prev_item(prev_values, prev_values_hash, key)
if prev_item then
prev_keys_still_present = prev_keys_still_present + 1
end

-- Deliberately leaves `changed` alone, so a reload that changed
-- nothing does not bump conf_version and rebuild every router.
-- Same semantics as sync_data, which re-runs the checker and filter
-- only for the keys that changed.
if prev_item and prev_item.modifiedIndex == item.modifiedIndex then
insert_tab(self.values, prev_item)
self.values_hash[key] = #self.values
self:upgrade_version(item.modifiedIndex)
goto continue
end

local data_valid = true
err = nil
if type(item.value) ~= "table" then
Expand Down Expand Up @@ -660,20 +663,26 @@ local function load_full_data(self, dir_res, headers, prev_values, prev_values_h
self.filter(item)
end

else
local prev_item = get_prev_item(prev_values, prev_values_hash, key)
if prev_item then
-- keep serving with the last valid configuration instead of
-- silently dropping the whole item on a full reload, see the
-- incremental path in sync_data for the same semantics
log.warn("failed to check item data of [", self.key, "/", key,
"], keep the previous configuration, err: ", err)
insert_tab(self.values, prev_item)
self.values_hash[key] = #self.values
end
elseif prev_item then
-- keep serving with the last valid configuration instead of
-- silently dropping the whole item on a full reload, see the
-- incremental path in sync_data for the same semantics
log.warn("failed to check item data of [", self.key, "/", key,
"], keep the previous configuration, err: ", err)
insert_tab(self.values, prev_item)
self.values_hash[key] = #self.values
end

self:upgrade_version(item.modifiedIndex)

::continue::
end

-- A deletion leaves every surviving key untouched, so it has to be
-- detected separately or a reload that only deletes would keep
-- serving the removed items.
if prev_values_hash and prev_keys_still_present < nkeys(prev_values_hash) then
changed = true
end
end

Expand Down
250 changes: 242 additions & 8 deletions t/core/config_etcd.t
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -521,7 +521,7 @@ main etcd watcher initialised, revision=



=== TEST 14: watch revision should be upgraded when timeout occurs
=== TEST 14: watch revision must not be upgraded when the watch times out
--- yaml_config
deployment:
role: traditional
Expand All @@ -538,6 +538,7 @@ nginx_config:
--- config
location /t {
content_by_lua_block {
local core = require("apisix.core")
local etcd = require("resty.etcd")
local etcd_cli, err = etcd.new({
http_host = "http://127.0.0.1:2379",
Expand All @@ -547,7 +548,8 @@ nginx_config:
return
end
ngx.sleep(2)
-- we will assert 4 lines of revision upgrade log because we have one worker and one privileged agent
-- write outside the watched prefix so that the global revision
-- moves on while the watch itself stays idle and keeps timing out
for i = 1, 2 do
local _, err = etcd_cli:set("/apache", "apisix")
if err then
Expand All @@ -556,17 +558,50 @@ nginx_config:
end
ngx.sleep(1)
end
ngx.say("passed")

-- The only other assertion here is that a log line is absent, which
-- would also hold if the watch were broken outright. So check delivery.
local _, err = etcd_cli:set("/apisix/routes/after-timeout", {
id = "after-timeout",
uri = "/after-timeout",
create_time = 1700000000,
update_time = 1700000000,
upstream = {type = "roundrobin", nodes = {["127.0.0.1:1980"] = 1}}
})
if err then
ngx.say("failed to set route: ", err)
return
end
-- polled, not slept: a fixed wait would just guess at CI latency
local delivered = false
for _ = 1, 25 do
ngx.sleep(0.2)
local obj = core.config.fetch_created_obj("/routes")
for _, item in ipairs(obj and obj.values or {}) do
if item and item.value and item.value.id == "after-timeout" then
delivered = true
break
end
end
if delivered then
break
end
end
ngx.say("update after timeout delivered: ", delivered)

local _, err = etcd_cli:delete("/apisix/routes/after-timeout")
if err then
ngx.log(ngx.WARN, "failed to clean up route: ", err)
end
}
}
--- timeout: 20
--- request
GET /t
--- response_body
passed
--- grep_error_log eval
qr/etcd watch timeout, upgrade revision to/
--- grep_error_log_out eval
qr/(etcd watch timeout, upgrade revision to\n){2,}/
update after timeout delivered: true
--- no_error_log
etcd watch timeout, upgrade revision to



Expand Down Expand Up @@ -862,3 +897,202 @@ GET /t
invalid new item loaded: false
--- no_error_log
keep the previous configuration



=== TEST 19: a full reload that changes nothing reuses the items and does not bump conf_version
--- timeout: 25
--- yaml_config
deployment:
role: traditional
role_traditional:
config_provider: etcd
etcd:
host:
- "http://127.0.0.1:2379"
prefix: /apisix
--- extra_yaml_config
nginx_config:
worker_processes: 1
--- config
location /t {
content_by_lua_block {
local core = require("apisix.core")
local etcd = require("resty.etcd")
local etcd_cli, err = etcd.new({
http_host = "http://127.0.0.1:2379",
})
if not etcd_cli then
ngx.say("failed to create etcd client: ", err)
return
end

local _, err = etcd_cli:set("/apisix/global_rules/1", {
id = "1",
create_time = 1700000000,
update_time = 1700000000,
plugins = {["response-rewrite"] = {headers = {set = {["X-T"] = "a"}}}}
})
if err then
ngx.say("failed to set global_rules/1: ", err)
return
end
ngx.sleep(2)

local obj = core.config.fetch_created_obj("/global_rules")
local before_version = obj.conf_version

-- A reload always builds a fresh `values` array, so losing this probe
-- proves it ran; the incremental path only mutates elements.
obj.values.array_probe = "old"
-- The items inside must survive: reusing them is the whole point.
for _, item in ipairs(obj.values) do
if item and item.value and item.value.id == "1" then
item.reload_probe = "kept"
end
end

-- Arm the recovery path taken after `compacted`. sync_data is parked
-- in waitdir, so the write below is what wakes it: /2 arrives
-- incrementally (+1), then the reload runs with /1 and /2 already in
-- memory at the revisions etcd reports, so it must not bump again.
obj.need_reload = true
local _, err = etcd_cli:set("/apisix/global_rules/2", {
id = "2",
create_time = 1700000000,
update_time = 1700000000,
plugins = {["response-rewrite"] = {headers = {set = {["X-T2"] = "b"}}}}
})
if err then
ngx.say("failed to set global_rules/2: ", err)
return
end
ngx.sleep(3)

local probe_kept = false
for _, item in ipairs(obj.values) do
if item and item.value and item.value.id == "1" then
probe_kept = (item.reload_probe == "kept")
end
end

ngx.say("reload ran: ", obj.values.array_probe == nil)
ngx.say("item reused: ", probe_kept)
ngx.say("conf_version bumped once, not twice: ",
obj.conf_version == before_version + 1)

for _, key in ipairs({"/apisix/global_rules/1", "/apisix/global_rules/2"}) do
local _, del_err = etcd_cli:delete(key)
if del_err then
ngx.log(ngx.WARN, "failed to clean up ", key, ": ", del_err)
end
end
ngx.sleep(1)
}
}
--- request
GET /t
--- response_body
reload ran: true
item reused: true
conf_version bumped once, not twice: true



=== TEST 20: a full reload that only deletes must still bump conf_version
--- timeout: 25
--- yaml_config
deployment:
role: traditional
role_traditional:
config_provider: etcd
etcd:
host:
- "http://127.0.0.1:2379"
prefix: /apisix
--- extra_yaml_config
nginx_config:
worker_processes: 1
--- config
location /t {
content_by_lua_block {
local core = require("apisix.core")
local etcd = require("resty.etcd")
local etcd_cli, err = etcd.new({
http_host = "http://127.0.0.1:2379",
})
if not etcd_cli then
ngx.say("failed to create etcd client: ", err)
return
end

local _, err = etcd_cli:set("/apisix/global_rules/1", {
id = "1",
create_time = 1700000000,
update_time = 1700000000,
plugins = {["response-rewrite"] = {headers = {set = {["X-T"] = "a"}}}}
})
if err then
ngx.say("failed to set global_rules/1: ", err)
return
end
ngx.sleep(2)

local obj = core.config.fetch_created_obj("/global_rules")
obj.values.array_probe = "old"

-- Live in memory, gone from etcd. Every surviving key is untouched
-- and therefore reused, so without an explicit deletion check
-- conf_version would not move and the routers would keep serving it.
local ghost = {
key = "/apisix/global_rules/ghost",
modifiedIndex = 1,
value = {id = "ghost", plugins = {}},
}
core.table.insert(obj.values, ghost)
obj.values_hash["ghost"] = #obj.values

local before_version = obj.conf_version

-- same wake-up mechanism as TEST 19: /2 arrives incrementally
-- (+1), then the reload drops the ghost (+1)
obj.need_reload = true
local _, err = etcd_cli:set("/apisix/global_rules/2", {
id = "2",
create_time = 1700000000,
update_time = 1700000000,
plugins = {["response-rewrite"] = {headers = {set = {["X-T2"] = "b"}}}}
})
if err then
ngx.say("failed to set global_rules/2: ", err)
return
end
ngx.sleep(3)

local found_ghost = false
for _, item in ipairs(obj.values) do
if item and item.value and item.value.id == "ghost" then
found_ghost = true
end
end

ngx.say("reload ran: ", obj.values.array_probe == nil)
ngx.say("ghost dropped: ", not found_ghost)
ngx.say("conf_version bumped for the deletion: ",
obj.conf_version == before_version + 2)

for _, key in ipairs({"/apisix/global_rules/1", "/apisix/global_rules/2"}) do
local _, del_err = etcd_cli:delete(key)
if del_err then
ngx.log(ngx.WARN, "failed to clean up ", key, ": ", del_err)
end
end
ngx.sleep(1)
}
}
--- request
GET /t
--- response_body
reload ran: true
ghost dropped: true
conf_version bumped for the deletion: true
Loading