diff --git a/kong/clustering/services/sync/hooks.lua b/kong/clustering/services/sync/hooks.lua index 7a3a1402558d..336bd3226aa1 100644 --- a/kong/clustering/services/sync/hooks.lua +++ b/kong/clustering/services/sync/hooks.lua @@ -76,10 +76,16 @@ end function _M:entity_delta_writer(row, name, options, ws_id, is_delete) + -- composite key, like { id = ... } + local schema = kong.db[name].schema + local pk = schema:extract_pk_values(row) + + assert(schema:validate_primary_key(pk)) + local deltas = { { type = name, - id = row.id, + pk = pk, ws_id = ws_id, row = is_delete and ngx_null or row, }, diff --git a/kong/clustering/services/sync/rpc.lua b/kong/clustering/services/sync/rpc.lua index 978a355fb1a2..625b394d3505 100644 --- a/kong/clustering/services/sync/rpc.lua +++ b/kong/clustering/services/sync/rpc.lua @@ -274,7 +274,7 @@ local function do_sync() else -- delete the entity - local old_entity, err = kong.db[delta_type]:select({ id = delta.id, }) -- TODO: composite key + local old_entity, err = kong.db[delta_type]:select(delta.pk) -- composite key if err then return nil, err end diff --git a/kong/clustering/services/sync/strategies/postgres.lua b/kong/clustering/services/sync/strategies/postgres.lua index 589b9f025047..616a4e38cc78 100644 --- a/kong/clustering/services/sync/strategies/postgres.lua +++ b/kong/clustering/services/sync/strategies/postgres.lua @@ -67,21 +67,21 @@ local NEW_VERSION_QUERY = [[ new_version integer; BEGIN INSERT INTO clustering_sync_version DEFAULT VALUES RETURNING version INTO new_version; - INSERT INTO clustering_sync_delta (version, type, id, ws_id, row) VALUES %s; + INSERT INTO clustering_sync_delta (version, type, pk, ws_id, row) VALUES %s; END $$; ]] -- deltas: { --- { type = "service", "id" = "d78eb00f-8702-4d6a-bfd9-e005f904ae3e", "ws_id" = "73478cf6-964f-412d-b1c4-8ac88d9e85e9", row = "JSON", } --- { type = "route", "id" = "0a5bac5c-b795-4981-95d2-919ba3390b7e", "ws_id" = "73478cf6-964f-412d-b1c4-8ac88d9e85e9", row = "JSON", } +-- { type = "service", "pk" = { id = "d78eb00f..." }, "ws_id" = "73478cf6...", row = "JSON", } +-- { type = "route", "pk" = { id = "0a5bac5c..." }, "ws_id" = "73478cf6...", row = "JSON", } -- } function _M:insert_delta(deltas) local buf = buffer.new() for _, d in ipairs(deltas) do buf:putf("(new_version, %s, %s, %s, %s)", self.connector:escape_literal(d.type), - self.connector:escape_literal(d.id), + self.connector:escape_literal(cjson_encode(d.pk)), self.connector:escape_literal(d.ws_id or kong.default_workspace), self.connector:escape_literal(cjson_encode(d.row))) end diff --git a/kong/db/migrations/core/024_370_to_380.lua b/kong/db/migrations/core/024_370_to_380.lua index 9d78807962cf..afb94a09f7a3 100644 --- a/kong/db/migrations/core/024_370_to_380.lua +++ b/kong/db/migrations/core/024_370_to_380.lua @@ -9,7 +9,7 @@ return { CREATE TABLE IF NOT EXISTS clustering_sync_delta ( "version" INT NOT NULL, "type" TEXT NOT NULL, - "id" UUID NOT NULL, + "pk" JSON NOT NULL, "ws_id" UUID NOT NULL, "row" JSON, FOREIGN KEY (version) REFERENCES clustering_sync_version(version) ON DELETE CASCADE diff --git a/spec/05-migration/db/migrations/core/024_370_to_380_spec.lua b/spec/05-migration/db/migrations/core/024_370_to_380_spec.lua index 0563c4c83f80..f4a6145d16be 100644 --- a/spec/05-migration/db/migrations/core/024_370_to_380_spec.lua +++ b/spec/05-migration/db/migrations/core/024_370_to_380_spec.lua @@ -10,7 +10,7 @@ describe("database migration", function() assert.database_has_relation("clustering_sync_delta") assert.table_has_column("clustering_sync_delta", "version", "integer") assert.table_has_column("clustering_sync_delta", "type", "text") - assert.table_has_column("clustering_sync_delta", "id", "uuid") + assert.table_has_column("clustering_sync_delta", "pk", "json") assert.table_has_column("clustering_sync_delta", "ws_id", "uuid") assert.table_has_column("clustering_sync_delta", "row", "json") end)