Skip to content
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

Jump to bottom

refactor(clustering/rpc): code clean of full sync in rpc #13823

Open
wants to merge 5 commits into
base: master
Choose a base branch
from
Open

refactor(clustering/rpc): code clean of full sync in rpc #13823

Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
08764e2
func full_sync_result()
chronolaw Nov 3, 2024
76dbe68
inc_sync_result
chronolaw Nov 3, 2024
6d2cdb7
debug logs
chronolaw Nov 3, 2024
8a54f71
todo comment
chronolaw Nov 3, 2024
3fdffb3
Revert "debug logs"
chronolaw Nov 3, 2024
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
45 changes: 24 additions & 21 deletions kong/clustering/services/sync/rpc.lua
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,13 +39,25 @@ function _M.new(strategy)
end


function _M:init_cp(manager)
local purge_delay = manager.conf.cluster_data_plane_purge_delay
local function inc_sync_result(res)
return { default = { deltas = res, wipe = false, }, }
end

local function gen_delta_result(res, wipe)
return { default = { deltas = res, wipe = wipe, }, }

local function full_sync_result()
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
local function full_sync_result()
local function full_sync()

The name is not very proper

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Talked offline, it is not a blocker.

local deltas, err = declarative.export_config_sync()
if not deltas then
return nil, err
end

-- wipe dp lmdb, full sync
return { default = { deltas = deltas, wipe = true, }, }
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
return { default = { deltas = deltas, wipe = true, }, }
return gen_delta_result(deltas, true)

Otherwise remove the function above

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Talked offline, it is not a blocker.

end


function _M:init_cp(manager)
local purge_delay = manager.conf.cluster_data_plane_purge_delay

-- CP
-- Method: kong.sync.v2.get_delta
-- Params: versions: list of current versions of the database
Expand Down Expand Up @@ -97,16 +109,11 @@ function _M:init_cp(manager)
", current_version: ", default_namespace_version,
", forcing a full sync")


local deltas, err = declarative.export_config_sync()
if not deltas then
return nil, err
end

-- wipe dp lmdb, full sync
return gen_delta_result(deltas, true)
return full_sync_result()
end

-- do we need an incremental sync?

local res, err = self.strategy:get_delta(default_namespace_version)
if not res then
return nil, err
Expand All @@ -117,13 +124,13 @@ function _M:init_cp(manager)
"[kong.sync.v2] no delta for node_id: ", node_id,
", current_version: ", default_namespace_version,
", node is already up to date" )
return gen_delta_result(res, false)
return inc_sync_result(res)
end

-- some deltas are returned, are they contiguous?
if res[1].version == default_namespace_version + 1 then
-- doesn't wipe dp lmdb, incremental sync
return gen_delta_result(res, false)
return inc_sync_result(res)
end

-- we need to full sync because holes are found
Expand All @@ -135,13 +142,7 @@ function _M:init_cp(manager)
", current_version: ", default_namespace_version,
", forcing a full sync")

local deltas, err = declarative.export_config_sync()
if not deltas then
return nil, err
end

-- wipe dp lmdb, full sync
return gen_delta_result(deltas, true)
return full_sync_result()
end)
end

Expand Down Expand Up @@ -364,6 +365,8 @@ end


function _M:sync_once(delay)
--- XXX TODO: check rpc connection is ready

return start_sync_timer(ngx.timer.at, delay or 0)
end

Expand Down
Loading