You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@apisix.apache.org by "ChuanFF (via GitHub)" <gi...@apache.org> on 2025/09/24 12:49:07 UTC
[PR] fix: correct handling of endpointSlices in Kubernetes service discovery [apisix]
ChuanFF opened a new pull request, #12634:
URL: https://github.com/apache/apisix/pull/12634
### Description
Currently, the Kubernetes service discovery implementation in APISIX incorrectly handles EndpointSlices, treating each individual slice as the complete set of endpoints for a service. This contradicts the official Kubernetes documentation.
According to the [Kubernetes EndpointSlice documentation](https://kubernetes.io/docs/reference/kubernetes-api/service-resources/endpoint-slice-v1/):
> "For a given service there may be multiple EndpointSlice objects which must be joined to produce the full set of endpoints; you can find all of the slices for a given service by listing EndpointSlices in the service's namespace whose `kubernetes.io/service-name` label contains the service's name."
The correct behavior should be to combine all EndpointSlices sharing the same `kubernetes.io/service-name` label to form the complete endpoint set for a service.
### Solution Approach
#### 1. Add EndpointSlices Cache
- Implemented `endpoint_slices_cache` to store all EndpointSlices with the structure:
`endpoint_slices_cache["k8s_id/namespace/k8s_service_name:port_name"] = {slice1, slice2, ...}`
- Any changes to EndpointSlices trigger updates to this cache
- The cache is then used to assemble the complete node list for each service, which is subsequently updated in shared memory
#### 2. Code Refactoring
- Renamed variables to better align with Kubernetes EndpointSlices data structure terminology
- Improved code readability and maintainability
#### 3. Some Bug Fixes
- **Fixed incorrect handling of `addresses` field**: The `addresses` field contains a list of IP strings, not objects with `.ip` attributes
- **Optimized node sorting logic**: Simplified the nested loop structure in `endpoint_buffer` processing:
```lua
-- Before (incorrect):
for _, ports in pairs(endpoint_buffer) do
for _, nodes in pairs(ports) do
core.table.sort(nodes, sort_nodes_cmp)
end
end
-- After (correct):
for _, nodes in pairs(endpoint_buffer) do
core.table.sort(nodes, sort_nodes_cmp)
end
```
The `endpoint_buffer` uses ports as keys and node lists as values, requiring only a single iteration level.
### Checklist
- [ ] I have explained the need for this PR and the problem it solves
- [ ] I have explained the changes or the new features added to this PR
- [ ] I have added tests corresponding to this change
- [ ] I have updated the documentation to reflect this change
- [ ] I have verified that this change is backward compatible (If not, please discuss on the [APISIX mailing list](https://github.com/apache/apisix/tree/master#community) first)
<!--
Note
1. Mark the PR as draft until it's ready to be reviewed.
2. Always add/update tests for any changes unless you have a good reason.
3. Always update the documentation to reflect the changes made in the PR.
4. Make a new commit to resolve conversations instead of `push -f`.
5. To resolve merge conflicts, merge master instead of rebasing.
6. Use "request review" to notify the reviewer after making changes.
7. Only a reviewer can mark a conversation as resolved.
-->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@apisix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix: correct handling of endpointSlices in Kubernetes service discovery [apisix]
Posted by "nic-6443 (via GitHub)" <gi...@apache.org>.
nic-6443 commented on code in PR #12634:
URL: https://github.com/apache/apisix/pull/12634#discussion_r2726561267
##########
apisix/discovery/kubernetes/init.lua:
##########
@@ -51,77 +51,172 @@ local function sort_nodes_cmp(left, right)
return left.port < right.port
end
-local function on_endpoint_slices_modified(handle, endpoint, operate)
- if handle.namespace_selector and
- not handle:namespace_selector(endpoint.metadata.namespace) then
- return
+local function update_endpoint_slices_cache(handle, endpoint_key, slice, slice_name)
+ if not handle.endpoint_slices_cache[endpoint_key] then
+ handle.endpoint_slices_cache[endpoint_key] = {}
end
+ local endpoint_slices = handle.endpoint_slices_cache[endpoint_key]
+ endpoint_slices[slice_name] = slice
+end
- core.log.debug(core.json.delay_encode(endpoint))
- core.table.clear(endpoint_buffer)
-
- local endpointslices = endpoint.endpoints
- if type(endpointslices) == "table" then
- for _, endpointslice in ipairs(endpointslices) do
- if endpointslice.addresses then
- local addresses = endpointslice.addresses
- for _, port in ipairs(endpoint.ports or {}) do
- local port_name
- if port.name then
- port_name = port.name
- elseif port.targetPort then
- port_name = tostring(port.targetPort)
- else
- port_name = tostring(port.port)
- end
-
- if endpointslice.conditions and endpointslice.conditions.ready then
- local nodes = endpoint_buffer[port_name]
- if nodes == nil then
- nodes = core.table.new(0, #endpointslices * #addresses)
- endpoint_buffer[port_name] = nodes
- end
-
- for _, address in ipairs(addresses) do
- core.table.insert(nodes, {
- host = address.ip,
- port = port.port,
- weight = handle.default_weight
- })
- end
- end
- end
+local function get_endpoints_from_cache(handle, endpoint_key)
+ local endpoint_slices = handle.endpoint_slices_cache[endpoint_key] or {}
+ local endpoints = {}
+ for _, endpoint_slice in pairs(endpoint_slices) do
+ for port, targets in pairs(endpoint_slice) do
+ if not endpoints[port] then
+ endpoints[port] = core.table.new(0, #targets)
end
+ core.table.insert_tail(endpoints[port], unpack(targets))
end
end
+ return endpoints
+end
- for _, ports in pairs(endpoint_buffer) do
- for _, nodes in pairs(ports) do
- core.table.sort(nodes, sort_nodes_cmp)
- end
- end
- local endpoint_key = endpoint.metadata.namespace .. "/" .. endpoint.metadata.name
- local endpoint_content = core.json.encode(endpoint_buffer, true)
+local function update_endpoint_dict(handle, endpoints, endpoint_key)
+ 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)
if err then
core.log.error("set endpoint version into discovery DICT failed, ", err)
- return
+ return false, err
end
_, err = handle.endpoint_dict:safe_set(endpoint_key, endpoint_content)
if err then
core.log.error("set endpoint into discovery DICT failed, ", err)
handle.endpoint_dict:delete(endpoint_key .. "#version")
+ return false, err
+ end
+end
+
+local function on_endpoint_slices_modified(handle, endpoint_slice, operate)
+ if not endpoint_slice.metadata then
+ core.log.error("endpoint_slice has no metadata, endpointSlice: ",
+ core.json.delay_encode(endpoint_slice))
+ return
+ end
+ if not endpoint_slice.metadata.name then
+ core.log.error("endpoint_slice has no metadata.name, endpointSlice: ",
+ core.json.delay_encode(endpoint_slice))
+ return
+ end
+ if not endpoint_slice.metadata.namespace then
+ core.log.error("endpoint_slice has no metadata.namespace, endpointSlice: ",
+ core.json.delay_encode(endpoint_slice))
+ return
+ end
+ if not endpoint_slice.metadata.labels
+ or not endpoint_slice.metadata.labels["kubernetes.io/service-name"] then
Review Comment:
Define `"kubernetes.io/service-name"` as a constant
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@apisix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix: correct handling of endpointSlices in Kubernetes service discovery [apisix]
Posted by "ChuanFF (via GitHub)" <gi...@apache.org>.
ChuanFF commented on code in PR #12634:
URL: https://github.com/apache/apisix/pull/12634#discussion_r2732595673
##########
apisix/discovery/kubernetes/init.lua:
##########
@@ -51,69 +52,133 @@ local function sort_nodes_cmp(left, right)
return left.port < right.port
end
-local function on_endpoint_slices_modified(handle, endpoint, operate)
- if handle.namespace_selector and
- not handle:namespace_selector(endpoint.metadata.namespace) then
- return
+local function update_endpoint_slices_cache(handle, endpoint_key, slice, slice_name)
+ if not handle.endpoint_slices_cache[endpoint_key] then
+ handle.endpoint_slices_cache[endpoint_key] = {}
end
+ local endpoint_slices = handle.endpoint_slices_cache[endpoint_key]
+ endpoint_slices[slice_name] = slice
+end
- core.log.debug(core.json.delay_encode(endpoint))
- core.table.clear(endpoint_buffer)
-
- local endpointslices = endpoint.endpoints
- if type(endpointslices) == "table" then
- for _, endpointslice in ipairs(endpointslices) do
- if endpointslice.addresses then
- local addresses = endpointslice.addresses
- for _, port in ipairs(endpoint.ports or {}) do
- local port_name
- if port.name then
- port_name = port.name
- elseif port.targetPort then
- port_name = tostring(port.targetPort)
- else
- port_name = tostring(port.port)
- end
-
- if endpointslice.conditions and endpointslice.conditions.ready then
- local nodes = endpoint_buffer[port_name]
- if nodes == nil then
- nodes = core.table.new(0, #endpointslices * #addresses)
- endpoint_buffer[port_name] = nodes
- end
-
- for _, address in ipairs(addresses) do
- core.table.insert(nodes, {
- host = address.ip,
- port = port.port,
- weight = handle.default_weight
- })
- end
- end
- end
+local function get_endpoints_from_cache(handle, endpoint_key)
+ local endpoint_slices = handle.endpoint_slices_cache[endpoint_key] or {}
+ local endpoints = {}
+ for _, endpoint_slice in pairs(endpoint_slices) do
+ for port, targets in pairs(endpoint_slice) do
+ if not endpoints[port] then
+ endpoints[port] = core.table.new(0, #targets)
end
+ core.table.insert_tail(endpoints[port], unpack(targets))
end
end
- for _, ports in pairs(endpoint_buffer) do
- for _, nodes in pairs(ports) do
- core.table.sort(nodes, sort_nodes_cmp)
- end
- end
- local endpoint_key = endpoint.metadata.namespace .. "/" .. endpoint.metadata.name
- local endpoint_content = core.json.encode(endpoint_buffer, true)
- local endpoint_version = ngx.crc32_long(endpoint_content)
+ return endpoints
+end
+local function update_endpoint_dict(handle, endpoints, endpoint_key)
+ 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)
if err then
- core.log.error("set endpoint version into discovery DICT failed, ", err)
- return
+ return false, "set endpoint version into discovery DICT failed, " .. err
end
_, err = handle.endpoint_dict:safe_set(endpoint_key, endpoint_content)
if err then
- core.log.error("set endpoint into discovery DICT failed, ", err)
handle.endpoint_dict:delete(endpoint_key .. "#version")
+ return false, "set endpoint into discovery DICT failed, " .. err
+ end
+
+ return true
+end
+
+local function validate_endpoint_slice(endpoint_slice)
+ if not endpoint_slice.metadata then
+ return false, "endpoint_slice has no metadata, endpointSlice: "
+ .. core.json.encode(endpoint_slice)
+ end
+ if not endpoint_slice.metadata.name then
+ return false, "endpoint_slice has no metadata.name, endpointSlice: "
+ .. core.json.encode(endpoint_slice)
+ end
+ if not endpoint_slice.metadata.namespace then
+ return false, "endpoint_slice has no metadata.namespace, endpointSlice: "
+ .. core.json.encode(endpoint_slice)
+ end
+ if not endpoint_slice.metadata.labels
+ or not endpoint_slice.metadata.labels[kubernetes_service_name_label] then
+ return false, "endpoint_slice has no service-name, endpointSlice: "
+ .. core.json.encode(endpoint_slice)
+ end
+
+ return true
+end
+
+local function on_endpoint_slices_modified(handle, endpoint_slice, operate)
+ local ok, err = validate_endpoint_slice(endpoint_slice)
+ if not ok then
+ core.log.error("endpoint_slice validation fail: ", err)
+ return
+ end
+ if handle.namespace_selector and
+ not handle:namespace_selector(endpoint_slice.metadata.namespace) then
+ return
+ end
+
+ core.log.debug(core.json.delay_encode(endpoint_slice))
Review Comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@apisix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix: correct handling of endpointSlices in Kubernetes service discovery [apisix]
Posted by "membphis (via GitHub)" <gi...@apache.org>.
membphis commented on code in PR #12634:
URL: https://github.com/apache/apisix/pull/12634#discussion_r2730495957
##########
apisix/discovery/kubernetes/init.lua:
##########
@@ -51,69 +52,133 @@ local function sort_nodes_cmp(left, right)
return left.port < right.port
end
-local function on_endpoint_slices_modified(handle, endpoint, operate)
- if handle.namespace_selector and
- not handle:namespace_selector(endpoint.metadata.namespace) then
- return
+local function update_endpoint_slices_cache(handle, endpoint_key, slice, slice_name)
+ if not handle.endpoint_slices_cache[endpoint_key] then
+ handle.endpoint_slices_cache[endpoint_key] = {}
end
+ local endpoint_slices = handle.endpoint_slices_cache[endpoint_key]
+ endpoint_slices[slice_name] = slice
+end
- core.log.debug(core.json.delay_encode(endpoint))
- core.table.clear(endpoint_buffer)
-
- local endpointslices = endpoint.endpoints
- if type(endpointslices) == "table" then
- for _, endpointslice in ipairs(endpointslices) do
- if endpointslice.addresses then
- local addresses = endpointslice.addresses
- for _, port in ipairs(endpoint.ports or {}) do
- local port_name
- if port.name then
- port_name = port.name
- elseif port.targetPort then
- port_name = tostring(port.targetPort)
- else
- port_name = tostring(port.port)
- end
-
- if endpointslice.conditions and endpointslice.conditions.ready then
- local nodes = endpoint_buffer[port_name]
- if nodes == nil then
- nodes = core.table.new(0, #endpointslices * #addresses)
- endpoint_buffer[port_name] = nodes
- end
-
- for _, address in ipairs(addresses) do
- core.table.insert(nodes, {
- host = address.ip,
- port = port.port,
- weight = handle.default_weight
- })
- end
- end
- end
+local function get_endpoints_from_cache(handle, endpoint_key)
+ local endpoint_slices = handle.endpoint_slices_cache[endpoint_key] or {}
+ local endpoints = {}
+ for _, endpoint_slice in pairs(endpoint_slices) do
+ for port, targets in pairs(endpoint_slice) do
+ if not endpoints[port] then
+ endpoints[port] = core.table.new(0, #targets)
end
+ core.table.insert_tail(endpoints[port], unpack(targets))
end
end
- for _, ports in pairs(endpoint_buffer) do
- for _, nodes in pairs(ports) do
- core.table.sort(nodes, sort_nodes_cmp)
- end
- end
- local endpoint_key = endpoint.metadata.namespace .. "/" .. endpoint.metadata.name
- local endpoint_content = core.json.encode(endpoint_buffer, true)
- local endpoint_version = ngx.crc32_long(endpoint_content)
+ return endpoints
+end
+local function update_endpoint_dict(handle, endpoints, endpoint_key)
+ 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)
if err then
- core.log.error("set endpoint version into discovery DICT failed, ", err)
- return
+ return false, "set endpoint version into discovery DICT failed, " .. err
end
_, err = handle.endpoint_dict:safe_set(endpoint_key, endpoint_content)
if err then
- core.log.error("set endpoint into discovery DICT failed, ", err)
handle.endpoint_dict:delete(endpoint_key .. "#version")
+ return false, "set endpoint into discovery DICT failed, " .. err
+ end
+
+ return true
+end
+
+local function validate_endpoint_slice(endpoint_slice)
+ if not endpoint_slice.metadata then
+ return false, "endpoint_slice has no metadata, endpointSlice: "
+ .. core.json.encode(endpoint_slice)
+ end
+ if not endpoint_slice.metadata.name then
+ return false, "endpoint_slice has no metadata.name, endpointSlice: "
+ .. core.json.encode(endpoint_slice)
+ end
+ if not endpoint_slice.metadata.namespace then
+ return false, "endpoint_slice has no metadata.namespace, endpointSlice: "
+ .. core.json.encode(endpoint_slice)
+ end
+ if not endpoint_slice.metadata.labels
+ or not endpoint_slice.metadata.labels[kubernetes_service_name_label] then
+ return false, "endpoint_slice has no service-name, endpointSlice: "
+ .. core.json.encode(endpoint_slice)
+ end
+
+ return true
+end
+
+local function on_endpoint_slices_modified(handle, endpoint_slice, operate)
+ local ok, err = validate_endpoint_slice(endpoint_slice)
+ if not ok then
+ core.log.error("endpoint_slice validation fail: ", err)
+ return
+ end
+ if handle.namespace_selector and
+ not handle:namespace_selector(endpoint_slice.metadata.namespace) then
+ return
+ end
+
+ core.log.debug(core.json.delay_encode(endpoint_slice))
Review Comment:
we can add more text to explain this error message
make it easier to understand
##########
apisix/discovery/kubernetes/init.lua:
##########
@@ -122,6 +187,36 @@ local function on_endpoint_slices_modified(handle, endpoint, operate)
end
end
+local function on_endpoint_slices_deleted(handle, endpoint_slice)
+ local ok, err = validate_endpoint_slice(endpoint_slice)
+ if not ok then
+ core.log.error("endpoint_slice validation fail: ", err)
+ return
+ end
+
+ if handle.namespace_selector and
+ not handle:namespace_selector(endpoint_slice.metadata.namespace) then
+ return
+ end
+
+ core.log.debug(core.json.delay_encode(endpoint_slice))
Review Comment:
ditto
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@apisix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix: correct handling of endpointSlices in Kubernetes service discovery [apisix]
Posted by "Baoyuantop (via GitHub)" <gi...@apache.org>.
Baoyuantop commented on PR #12634:
URL: https://github.com/apache/apisix/pull/12634#issuecomment-3689357212
Hi @ChuanFF, you can resolve the conflict by merging the code in the main branch.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@apisix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix: correct handling of endpointSlices in Kubernetes service discovery [apisix]
Posted by "ChuanFF (via GitHub)" <gi...@apache.org>.
ChuanFF commented on code in PR #12634:
URL: https://github.com/apache/apisix/pull/12634#discussion_r2727642038
##########
apisix/discovery/kubernetes/init.lua:
##########
@@ -51,77 +51,172 @@ local function sort_nodes_cmp(left, right)
return left.port < right.port
end
-local function on_endpoint_slices_modified(handle, endpoint, operate)
- if handle.namespace_selector and
- not handle:namespace_selector(endpoint.metadata.namespace) then
- return
+local function update_endpoint_slices_cache(handle, endpoint_key, slice, slice_name)
+ if not handle.endpoint_slices_cache[endpoint_key] then
+ handle.endpoint_slices_cache[endpoint_key] = {}
end
+ local endpoint_slices = handle.endpoint_slices_cache[endpoint_key]
+ endpoint_slices[slice_name] = slice
+end
- core.log.debug(core.json.delay_encode(endpoint))
- core.table.clear(endpoint_buffer)
-
- local endpointslices = endpoint.endpoints
- if type(endpointslices) == "table" then
- for _, endpointslice in ipairs(endpointslices) do
- if endpointslice.addresses then
- local addresses = endpointslice.addresses
- for _, port in ipairs(endpoint.ports or {}) do
- local port_name
- if port.name then
- port_name = port.name
- elseif port.targetPort then
- port_name = tostring(port.targetPort)
- else
- port_name = tostring(port.port)
- end
-
- if endpointslice.conditions and endpointslice.conditions.ready then
- local nodes = endpoint_buffer[port_name]
- if nodes == nil then
- nodes = core.table.new(0, #endpointslices * #addresses)
- endpoint_buffer[port_name] = nodes
- end
-
- for _, address in ipairs(addresses) do
- core.table.insert(nodes, {
- host = address.ip,
- port = port.port,
- weight = handle.default_weight
- })
- end
- end
- end
+local function get_endpoints_from_cache(handle, endpoint_key)
+ local endpoint_slices = handle.endpoint_slices_cache[endpoint_key] or {}
+ local endpoints = {}
+ for _, endpoint_slice in pairs(endpoint_slices) do
+ for port, targets in pairs(endpoint_slice) do
+ if not endpoints[port] then
+ endpoints[port] = core.table.new(0, #targets)
end
+ core.table.insert_tail(endpoints[port], unpack(targets))
end
end
+ return endpoints
+end
- for _, ports in pairs(endpoint_buffer) do
- for _, nodes in pairs(ports) do
- core.table.sort(nodes, sort_nodes_cmp)
- end
- end
- local endpoint_key = endpoint.metadata.namespace .. "/" .. endpoint.metadata.name
- local endpoint_content = core.json.encode(endpoint_buffer, true)
+local function update_endpoint_dict(handle, endpoints, endpoint_key)
+ 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)
if err then
core.log.error("set endpoint version into discovery DICT failed, ", err)
- return
+ return false, err
Review Comment:
done, i check the return value at the call site; if it's false, I log an error and return.
##########
apisix/discovery/kubernetes/init.lua:
##########
@@ -51,77 +51,172 @@ local function sort_nodes_cmp(left, right)
return left.port < right.port
end
-local function on_endpoint_slices_modified(handle, endpoint, operate)
- if handle.namespace_selector and
- not handle:namespace_selector(endpoint.metadata.namespace) then
- return
+local function update_endpoint_slices_cache(handle, endpoint_key, slice, slice_name)
+ if not handle.endpoint_slices_cache[endpoint_key] then
+ handle.endpoint_slices_cache[endpoint_key] = {}
end
+ local endpoint_slices = handle.endpoint_slices_cache[endpoint_key]
+ endpoint_slices[slice_name] = slice
+end
- core.log.debug(core.json.delay_encode(endpoint))
- core.table.clear(endpoint_buffer)
-
- local endpointslices = endpoint.endpoints
- if type(endpointslices) == "table" then
- for _, endpointslice in ipairs(endpointslices) do
- if endpointslice.addresses then
- local addresses = endpointslice.addresses
- for _, port in ipairs(endpoint.ports or {}) do
- local port_name
- if port.name then
- port_name = port.name
- elseif port.targetPort then
- port_name = tostring(port.targetPort)
- else
- port_name = tostring(port.port)
- end
-
- if endpointslice.conditions and endpointslice.conditions.ready then
- local nodes = endpoint_buffer[port_name]
- if nodes == nil then
- nodes = core.table.new(0, #endpointslices * #addresses)
- endpoint_buffer[port_name] = nodes
- end
-
- for _, address in ipairs(addresses) do
- core.table.insert(nodes, {
- host = address.ip,
- port = port.port,
- weight = handle.default_weight
- })
- end
- end
- end
+local function get_endpoints_from_cache(handle, endpoint_key)
+ local endpoint_slices = handle.endpoint_slices_cache[endpoint_key] or {}
+ local endpoints = {}
+ for _, endpoint_slice in pairs(endpoint_slices) do
+ for port, targets in pairs(endpoint_slice) do
+ if not endpoints[port] then
+ endpoints[port] = core.table.new(0, #targets)
end
+ core.table.insert_tail(endpoints[port], unpack(targets))
end
end
+ return endpoints
+end
- for _, ports in pairs(endpoint_buffer) do
- for _, nodes in pairs(ports) do
- core.table.sort(nodes, sort_nodes_cmp)
- end
- end
- local endpoint_key = endpoint.metadata.namespace .. "/" .. endpoint.metadata.name
- local endpoint_content = core.json.encode(endpoint_buffer, true)
+local function update_endpoint_dict(handle, endpoints, endpoint_key)
+ 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)
if err then
core.log.error("set endpoint version into discovery DICT failed, ", err)
- return
+ return false, err
end
_, err = handle.endpoint_dict:safe_set(endpoint_key, endpoint_content)
if err then
core.log.error("set endpoint into discovery DICT failed, ", err)
handle.endpoint_dict:delete(endpoint_key .. "#version")
+ return false, err
+ end
+end
+
+local function on_endpoint_slices_modified(handle, endpoint_slice, operate)
+ if not endpoint_slice.metadata then
+ core.log.error("endpoint_slice has no metadata, endpointSlice: ",
+ core.json.delay_encode(endpoint_slice))
+ return
+ end
+ if not endpoint_slice.metadata.name then
+ core.log.error("endpoint_slice has no metadata.name, endpointSlice: ",
+ core.json.delay_encode(endpoint_slice))
+ return
+ end
+ if not endpoint_slice.metadata.namespace then
+ core.log.error("endpoint_slice has no metadata.namespace, endpointSlice: ",
+ core.json.delay_encode(endpoint_slice))
+ return
+ end
+ if not endpoint_slice.metadata.labels
+ or not endpoint_slice.metadata.labels["kubernetes.io/service-name"] then
Review Comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@apisix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix: correct handling of endpointSlices in Kubernetes service discovery [apisix]
Posted by "Baoyuantop (via GitHub)" <gi...@apache.org>.
Baoyuantop merged PR #12634:
URL: https://github.com/apache/apisix/pull/12634
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@apisix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix: correct handling of endpointSlices in Kubernetes service discovery [apisix]
Posted by "ChuanFF (via GitHub)" <gi...@apache.org>.
ChuanFF commented on code in PR #12634:
URL: https://github.com/apache/apisix/pull/12634#discussion_r2727643776
##########
apisix/discovery/kubernetes/init.lua:
##########
@@ -51,77 +51,172 @@ local function sort_nodes_cmp(left, right)
return left.port < right.port
end
-local function on_endpoint_slices_modified(handle, endpoint, operate)
- if handle.namespace_selector and
- not handle:namespace_selector(endpoint.metadata.namespace) then
- return
+local function update_endpoint_slices_cache(handle, endpoint_key, slice, slice_name)
+ if not handle.endpoint_slices_cache[endpoint_key] then
+ handle.endpoint_slices_cache[endpoint_key] = {}
end
+ local endpoint_slices = handle.endpoint_slices_cache[endpoint_key]
+ endpoint_slices[slice_name] = slice
+end
- core.log.debug(core.json.delay_encode(endpoint))
- core.table.clear(endpoint_buffer)
-
- local endpointslices = endpoint.endpoints
- if type(endpointslices) == "table" then
- for _, endpointslice in ipairs(endpointslices) do
- if endpointslice.addresses then
- local addresses = endpointslice.addresses
- for _, port in ipairs(endpoint.ports or {}) do
- local port_name
- if port.name then
- port_name = port.name
- elseif port.targetPort then
- port_name = tostring(port.targetPort)
- else
- port_name = tostring(port.port)
- end
-
- if endpointslice.conditions and endpointslice.conditions.ready then
- local nodes = endpoint_buffer[port_name]
- if nodes == nil then
- nodes = core.table.new(0, #endpointslices * #addresses)
- endpoint_buffer[port_name] = nodes
- end
-
- for _, address in ipairs(addresses) do
- core.table.insert(nodes, {
- host = address.ip,
- port = port.port,
- weight = handle.default_weight
- })
- end
- end
- end
+local function get_endpoints_from_cache(handle, endpoint_key)
+ local endpoint_slices = handle.endpoint_slices_cache[endpoint_key] or {}
+ local endpoints = {}
+ for _, endpoint_slice in pairs(endpoint_slices) do
+ for port, targets in pairs(endpoint_slice) do
+ if not endpoints[port] then
+ endpoints[port] = core.table.new(0, #targets)
end
+ core.table.insert_tail(endpoints[port], unpack(targets))
end
end
+ return endpoints
+end
- for _, ports in pairs(endpoint_buffer) do
- for _, nodes in pairs(ports) do
- core.table.sort(nodes, sort_nodes_cmp)
- end
- end
- local endpoint_key = endpoint.metadata.namespace .. "/" .. endpoint.metadata.name
- local endpoint_content = core.json.encode(endpoint_buffer, true)
+local function update_endpoint_dict(handle, endpoints, endpoint_key)
+ 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)
if err then
core.log.error("set endpoint version into discovery DICT failed, ", err)
- return
+ return false, err
end
_, err = handle.endpoint_dict:safe_set(endpoint_key, endpoint_content)
if err then
core.log.error("set endpoint into discovery DICT failed, ", err)
handle.endpoint_dict:delete(endpoint_key .. "#version")
+ return false, err
+ end
+end
+
+local function on_endpoint_slices_modified(handle, endpoint_slice, operate)
+ if not endpoint_slice.metadata then
+ core.log.error("endpoint_slice has no metadata, endpointSlice: ",
+ core.json.delay_encode(endpoint_slice))
+ return
+ end
+ if not endpoint_slice.metadata.name then
+ core.log.error("endpoint_slice has no metadata.name, endpointSlice: ",
+ core.json.delay_encode(endpoint_slice))
+ return
+ end
+ if not endpoint_slice.metadata.namespace then
+ core.log.error("endpoint_slice has no metadata.namespace, endpointSlice: ",
+ core.json.delay_encode(endpoint_slice))
+ return
+ end
+ if not endpoint_slice.metadata.labels
+ or not endpoint_slice.metadata.labels["kubernetes.io/service-name"] then
+ core.log.error("endpoint_slice has no service-name, endpointSlice: ",
+ core.json.delay_encode(endpoint_slice))
return
end
+
+ if handle.namespace_selector and
+ not handle:namespace_selector(endpoint_slice.metadata.namespace) then
+ return
+ end
+
+ core.log.debug(core.json.delay_encode(endpoint_slice))
+ --record nodes to every port in service
+ local port_to_nodes = {}
+
+ local slice_endpoints = endpoint_slice.endpoints
+ if not slice_endpoints or slice_endpoints == ngx.null then
+ slice_endpoints = {}
+ end
+
+ for _, endpoint in ipairs(slice_endpoints) do
+ if endpoint.addresses
+ and endpoint.conditions
+ and endpoint.conditions.ready then
+ local addresses = endpoint.addresses
+ for _, port in ipairs(endpoint_slice.ports or {}) do
+ local port_name
+ if port.name then
+ port_name = port.name
+ elseif port.targetPort then
+ port_name = tostring(port.targetPort)
+ else
+ port_name = tostring(port.port)
+ end
+
+ local nodes = port_to_nodes[port_name]
+ if nodes == nil then
+ nodes = core.table.new(0, #slice_endpoints * #addresses)
+ port_to_nodes[port_name] = nodes
+ end
+
+ for _, ip in ipairs(addresses) do
+ core.table.insert(nodes, {
+ host = ip,
+ port = port.port,
+ weight = handle.default_weight
+ })
+ end
+ end
+ end
+ end
+
+ local endpoint_key = endpoint_slice.metadata.namespace
+ .. "/" .. endpoint_slice.metadata.labels["kubernetes.io/service-name"]
+ update_endpoint_slices_cache(handle, endpoint_key, port_to_nodes, endpoint_slice.metadata.name)
+
+ local cached_endpoints = get_endpoints_from_cache(handle, endpoint_key)
+ for _, nodes in pairs(cached_endpoints) do
+ core.table.sort(nodes, sort_nodes_cmp)
+ end
+
+ update_endpoint_dict(handle, cached_endpoints, endpoint_key)
if operate == "list" then
handle.current_keys_hash[endpoint_key] = true
handle.current_keys_hash[endpoint_key .. "#version"] = true
end
end
+local function on_endpoint_slices_deleted(handle, endpoint_slice)
+ if not endpoint_slice.metadata then
+ core.log.error("endpoint_slice has no metadata, endpointSlice: ",
+ core.json.delay_encode(endpoint_slice))
+ return
+ end
+ if not endpoint_slice.metadata.name then
+ core.log.error("endpoint_slice has no metadata.name, endpointSlice: ",
+ core.json.delay_encode(endpoint_slice))
+ return
+ end
+ if not endpoint_slice.metadata.namespace then
+ core.log.error("endpoint_slice has no metadata.namespace, endpointSlice: ",
+ core.json.delay_encode(endpoint_slice))
+ return
+ end
+ if not endpoint_slice.metadata.labels
+ or not endpoint_slice.metadata.labels["kubernetes.io/service-name"] then
+ core.log.error("endpoint_slice has no service-name, endpointSlice: ",
+ core.json.delay_encode(endpoint_slice))
+ return
Review Comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@apisix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix: correct handling of endpointSlices in Kubernetes service discovery [apisix]
Posted by "ChuanFF (via GitHub)" <gi...@apache.org>.
ChuanFF commented on PR #12634:
URL: https://github.com/apache/apisix/pull/12634#issuecomment-3799663003
The code changes have been completed; please help review the code again.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@apisix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix: correct handling of endpointSlices in Kubernetes service discovery [apisix]
Posted by "nic-6443 (via GitHub)" <gi...@apache.org>.
nic-6443 commented on code in PR #12634:
URL: https://github.com/apache/apisix/pull/12634#discussion_r2726555157
##########
apisix/discovery/kubernetes/init.lua:
##########
@@ -51,77 +51,172 @@ local function sort_nodes_cmp(left, right)
return left.port < right.port
end
-local function on_endpoint_slices_modified(handle, endpoint, operate)
- if handle.namespace_selector and
- not handle:namespace_selector(endpoint.metadata.namespace) then
- return
+local function update_endpoint_slices_cache(handle, endpoint_key, slice, slice_name)
+ if not handle.endpoint_slices_cache[endpoint_key] then
+ handle.endpoint_slices_cache[endpoint_key] = {}
end
+ local endpoint_slices = handle.endpoint_slices_cache[endpoint_key]
+ endpoint_slices[slice_name] = slice
+end
- core.log.debug(core.json.delay_encode(endpoint))
- core.table.clear(endpoint_buffer)
-
- local endpointslices = endpoint.endpoints
- if type(endpointslices) == "table" then
- for _, endpointslice in ipairs(endpointslices) do
- if endpointslice.addresses then
- local addresses = endpointslice.addresses
- for _, port in ipairs(endpoint.ports or {}) do
- local port_name
- if port.name then
- port_name = port.name
- elseif port.targetPort then
- port_name = tostring(port.targetPort)
- else
- port_name = tostring(port.port)
- end
-
- if endpointslice.conditions and endpointslice.conditions.ready then
- local nodes = endpoint_buffer[port_name]
- if nodes == nil then
- nodes = core.table.new(0, #endpointslices * #addresses)
- endpoint_buffer[port_name] = nodes
- end
-
- for _, address in ipairs(addresses) do
- core.table.insert(nodes, {
- host = address.ip,
- port = port.port,
- weight = handle.default_weight
- })
- end
- end
- end
+local function get_endpoints_from_cache(handle, endpoint_key)
+ local endpoint_slices = handle.endpoint_slices_cache[endpoint_key] or {}
+ local endpoints = {}
+ for _, endpoint_slice in pairs(endpoint_slices) do
+ for port, targets in pairs(endpoint_slice) do
+ if not endpoints[port] then
+ endpoints[port] = core.table.new(0, #targets)
end
+ core.table.insert_tail(endpoints[port], unpack(targets))
end
end
+ return endpoints
+end
- for _, ports in pairs(endpoint_buffer) do
- for _, nodes in pairs(ports) do
- core.table.sort(nodes, sort_nodes_cmp)
- end
- end
- local endpoint_key = endpoint.metadata.namespace .. "/" .. endpoint.metadata.name
- local endpoint_content = core.json.encode(endpoint_buffer, true)
+local function update_endpoint_dict(handle, endpoints, endpoint_key)
+ 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)
if err then
core.log.error("set endpoint version into discovery DICT failed, ", err)
- return
+ return false, err
Review Comment:
This return value is not used. Additionally, when this function executes normally, there is a missing return value.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@apisix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix: correct handling of endpointSlices in Kubernetes service discovery [apisix]
Posted by "nic-6443 (via GitHub)" <gi...@apache.org>.
nic-6443 commented on code in PR #12634:
URL: https://github.com/apache/apisix/pull/12634#discussion_r2726563312
##########
apisix/discovery/kubernetes/init.lua:
##########
@@ -51,77 +51,172 @@ local function sort_nodes_cmp(left, right)
return left.port < right.port
end
-local function on_endpoint_slices_modified(handle, endpoint, operate)
- if handle.namespace_selector and
- not handle:namespace_selector(endpoint.metadata.namespace) then
- return
+local function update_endpoint_slices_cache(handle, endpoint_key, slice, slice_name)
+ if not handle.endpoint_slices_cache[endpoint_key] then
+ handle.endpoint_slices_cache[endpoint_key] = {}
end
+ local endpoint_slices = handle.endpoint_slices_cache[endpoint_key]
+ endpoint_slices[slice_name] = slice
+end
- core.log.debug(core.json.delay_encode(endpoint))
- core.table.clear(endpoint_buffer)
-
- local endpointslices = endpoint.endpoints
- if type(endpointslices) == "table" then
- for _, endpointslice in ipairs(endpointslices) do
- if endpointslice.addresses then
- local addresses = endpointslice.addresses
- for _, port in ipairs(endpoint.ports or {}) do
- local port_name
- if port.name then
- port_name = port.name
- elseif port.targetPort then
- port_name = tostring(port.targetPort)
- else
- port_name = tostring(port.port)
- end
-
- if endpointslice.conditions and endpointslice.conditions.ready then
- local nodes = endpoint_buffer[port_name]
- if nodes == nil then
- nodes = core.table.new(0, #endpointslices * #addresses)
- endpoint_buffer[port_name] = nodes
- end
-
- for _, address in ipairs(addresses) do
- core.table.insert(nodes, {
- host = address.ip,
- port = port.port,
- weight = handle.default_weight
- })
- end
- end
- end
+local function get_endpoints_from_cache(handle, endpoint_key)
+ local endpoint_slices = handle.endpoint_slices_cache[endpoint_key] or {}
+ local endpoints = {}
+ for _, endpoint_slice in pairs(endpoint_slices) do
+ for port, targets in pairs(endpoint_slice) do
+ if not endpoints[port] then
+ endpoints[port] = core.table.new(0, #targets)
end
+ core.table.insert_tail(endpoints[port], unpack(targets))
end
end
+ return endpoints
+end
- for _, ports in pairs(endpoint_buffer) do
- for _, nodes in pairs(ports) do
- core.table.sort(nodes, sort_nodes_cmp)
- end
- end
- local endpoint_key = endpoint.metadata.namespace .. "/" .. endpoint.metadata.name
- local endpoint_content = core.json.encode(endpoint_buffer, true)
+local function update_endpoint_dict(handle, endpoints, endpoint_key)
+ 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)
if err then
core.log.error("set endpoint version into discovery DICT failed, ", err)
- return
+ return false, err
end
_, err = handle.endpoint_dict:safe_set(endpoint_key, endpoint_content)
if err then
core.log.error("set endpoint into discovery DICT failed, ", err)
handle.endpoint_dict:delete(endpoint_key .. "#version")
+ return false, err
+ end
+end
+
+local function on_endpoint_slices_modified(handle, endpoint_slice, operate)
+ if not endpoint_slice.metadata then
+ core.log.error("endpoint_slice has no metadata, endpointSlice: ",
+ core.json.delay_encode(endpoint_slice))
+ return
+ end
+ if not endpoint_slice.metadata.name then
+ core.log.error("endpoint_slice has no metadata.name, endpointSlice: ",
+ core.json.delay_encode(endpoint_slice))
+ return
+ end
+ if not endpoint_slice.metadata.namespace then
+ core.log.error("endpoint_slice has no metadata.namespace, endpointSlice: ",
+ core.json.delay_encode(endpoint_slice))
+ return
+ end
+ if not endpoint_slice.metadata.labels
+ or not endpoint_slice.metadata.labels["kubernetes.io/service-name"] then
+ core.log.error("endpoint_slice has no service-name, endpointSlice: ",
+ core.json.delay_encode(endpoint_slice))
return
end
+
+ if handle.namespace_selector and
+ not handle:namespace_selector(endpoint_slice.metadata.namespace) then
+ return
+ end
+
+ core.log.debug(core.json.delay_encode(endpoint_slice))
+ --record nodes to every port in service
+ local port_to_nodes = {}
+
+ local slice_endpoints = endpoint_slice.endpoints
+ if not slice_endpoints or slice_endpoints == ngx.null then
+ slice_endpoints = {}
+ end
+
+ for _, endpoint in ipairs(slice_endpoints) do
+ if endpoint.addresses
+ and endpoint.conditions
+ and endpoint.conditions.ready then
+ local addresses = endpoint.addresses
+ for _, port in ipairs(endpoint_slice.ports or {}) do
+ local port_name
+ if port.name then
+ port_name = port.name
+ elseif port.targetPort then
+ port_name = tostring(port.targetPort)
+ else
+ port_name = tostring(port.port)
+ end
+
+ local nodes = port_to_nodes[port_name]
+ if nodes == nil then
+ nodes = core.table.new(0, #slice_endpoints * #addresses)
+ port_to_nodes[port_name] = nodes
+ end
+
+ for _, ip in ipairs(addresses) do
+ core.table.insert(nodes, {
+ host = ip,
+ port = port.port,
+ weight = handle.default_weight
+ })
+ end
+ end
+ end
+ end
+
+ local endpoint_key = endpoint_slice.metadata.namespace
+ .. "/" .. endpoint_slice.metadata.labels["kubernetes.io/service-name"]
+ update_endpoint_slices_cache(handle, endpoint_key, port_to_nodes, endpoint_slice.metadata.name)
+
+ local cached_endpoints = get_endpoints_from_cache(handle, endpoint_key)
+ for _, nodes in pairs(cached_endpoints) do
+ core.table.sort(nodes, sort_nodes_cmp)
+ end
+
+ update_endpoint_dict(handle, cached_endpoints, endpoint_key)
if operate == "list" then
handle.current_keys_hash[endpoint_key] = true
handle.current_keys_hash[endpoint_key .. "#version"] = true
end
end
+local function on_endpoint_slices_deleted(handle, endpoint_slice)
+ if not endpoint_slice.metadata then
+ core.log.error("endpoint_slice has no metadata, endpointSlice: ",
+ core.json.delay_encode(endpoint_slice))
+ return
+ end
+ if not endpoint_slice.metadata.name then
+ core.log.error("endpoint_slice has no metadata.name, endpointSlice: ",
+ core.json.delay_encode(endpoint_slice))
+ return
+ end
+ if not endpoint_slice.metadata.namespace then
+ core.log.error("endpoint_slice has no metadata.namespace, endpointSlice: ",
+ core.json.delay_encode(endpoint_slice))
+ return
+ end
+ if not endpoint_slice.metadata.labels
+ or not endpoint_slice.metadata.labels["kubernetes.io/service-name"] then
+ core.log.error("endpoint_slice has no service-name, endpointSlice: ",
+ core.json.delay_encode(endpoint_slice))
+ return
Review Comment:
This part of the code is completely duplicated with that in `on_endpoint_slices_modified` and can be extracted into an independent function.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@apisix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix: correct handling of endpointSlices in Kubernetes service discovery [apisix]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #12634:
URL: https://github.com/apache/apisix/pull/12634#issuecomment-3569895291
This pull request has been marked as stale due to 60 days of inactivity. It will be closed in 4 weeks if no further activity occurs. If you think that's incorrect or this pull request should instead be reviewed, please simply write any comment. Even if closed, you can still revive the PR at any time or discuss it on the dev@apisix.apache.org list. Thank you for your contributions.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@apisix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org