zlei6/go-zero
1
0
Fork 0
mirror of https://github.com/zeromicro/go-zero.git synced 2026-05-07 15:10:01 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(zrpc): migrate kube resolver from Endpoints to EndpointSlice API (#4987)

Browse Source 
Signed-off-by: soasurs <soasurs@gmail.com>
Co-authored-by: Kevin Wan <wanjunfeng@gmail.com>
This commit is contained in:
soasurs
2025-12-11 23:09:08 +08:00
committed by GitHub
parent 858f8ca82e
commit 3d291328d8
4 changed files with 180 additions and 196 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
.github/workflows/reviewdog.yml vendored
View File

@@ -6,6 +6,11 @@ jobs:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v6
- uses: actions/setup-go@v6
with:
go-version-file: go.mod
check-latest: true
cache: true
- uses: reviewdog/action-staticcheck@v1
with:
github_token: ${{ secrets.github_token }}

46
zrpc/resolver/internal/kube/eventhandler.go
View File

@@ -5,7 +5,7 @@ import (
"github.com/zeromicro/go-zero/core/lang"
"github.com/zeromicro/go-zero/core/logx"
"k8s.io/api/core/v1"
"k8s.io/api/discovery/v1"
"k8s.io/client-go/tools/cache"
)
@@ -28,9 +28,9 @@ func NewEventHandler(update func([]string)) *EventHandler {
// OnAdd handles the endpoints add events.
func (h *EventHandler) OnAdd(obj any, _ bool) {
endpoints, ok := obj.(*v1.Endpoints)
endpoints, ok := obj.(*v1.EndpointSlice)
if !ok {
logx.Errorf("%v is not an object with type *v1.Endpoints", obj)
logx.Errorf("%v is not an object with type *v1.EndpointSlice", obj)
return
}
@@ -38,10 +38,10 @@ func (h *EventHandler) OnAdd(obj any, _ bool) {
defer h.lock.Unlock()
var changed bool
for _, sub := range endpoints.Subsets {
for _, point := range sub.Addresses {
if _, ok := h.endpoints[point.IP]; !ok {
h.endpoints[point.IP] = lang.Placeholder
for _, point := range endpoints.Endpoints {
for _, address := range point.Addresses {
if _, ok := h.endpoints[address]; !ok {
h.endpoints[address] = lang.Placeholder
changed = true
}
}
@@ -54,9 +54,9 @@ func (h *EventHandler) OnAdd(obj any, _ bool) {
// OnDelete handles the endpoints delete events.
func (h *EventHandler) OnDelete(obj any) {
endpoints, ok := obj.(*v1.Endpoints)
endpoints, ok := obj.(*v1.EndpointSlice)
if !ok {
logx.Errorf("%v is not an object with type *v1.Endpoints", obj)
logx.Errorf("%v is not an object with type *v1.EndpointSlice", obj)
return
}
@@ -64,10 +64,10 @@ func (h *EventHandler) OnDelete(obj any) {
defer h.lock.Unlock()
var changed bool
for _, sub := range endpoints.Subsets {
for _, point := range sub.Addresses {
if _, ok := h.endpoints[point.IP]; ok {
delete(h.endpoints, point.IP)
for _, point := range endpoints.Endpoints {
for _, address := range point.Addresses {
if _, ok := h.endpoints[address]; ok {
delete(h.endpoints, address)
changed = true
}
}
@@ -80,35 +80,35 @@ func (h *EventHandler) OnDelete(obj any) {
// OnUpdate handles the endpoints update events.
func (h *EventHandler) OnUpdate(oldObj, newObj any) {
oldEndpoints, ok := oldObj.(*v1.Endpoints)
oldEndpointSlices, ok := oldObj.(*v1.EndpointSlice)
if !ok {
logx.Errorf("%v is not an object with type *v1.Endpoints", oldObj)
logx.Errorf("%v is not an object with type *v1.EndpointSlice", oldObj)
return
}
newEndpoints, ok := newObj.(*v1.Endpoints)
newEndpointSlices, ok := newObj.(*v1.EndpointSlice)
if !ok {
logx.Errorf("%v is not an object with type *v1.Endpoints", newObj)
logx.Errorf("%v is not an object with type *v1.EndpointSlice", newObj)
return
}
if oldEndpoints.ResourceVersion == newEndpoints.ResourceVersion {
if oldEndpointSlices.ResourceVersion == newEndpointSlices.ResourceVersion {
return
}
h.Update(newEndpoints)
h.Update(newEndpointSlices)
}
// Update updates the endpoints.
func (h *EventHandler) Update(endpoints *v1.Endpoints) {
func (h *EventHandler) Update(endpoints *v1.EndpointSlice) {
h.lock.Lock()
defer h.lock.Unlock()
old := h.endpoints
h.endpoints = make(map[string]lang.PlaceholderType)
for _, sub := range endpoints.Subsets {
for _, point := range sub.Addresses {
h.endpoints[point.IP] = lang.Placeholder
for _, point := range endpoints.Endpoints {
for _, address := range point.Addresses {
h.endpoints[address] = lang.Placeholder
}
}

274
zrpc/resolver/internal/kube/eventhandler_test.go
View File

@@ -4,7 +4,7 @@ import (
"testing"
"github.com/stretchr/testify/assert"
v1 "k8s.io/api/core/v1"
discoveryv1 "k8s.io/api/discovery/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)
@@ -14,21 +14,19 @@ func TestAdd(t *testing.T) {
endpoints = change
})
h.OnAdd("bad", false)
h.OnAdd(&v1.Endpoints{Subsets: []v1.EndpointSubset{
{
Addresses: []v1.EndpointAddress{
{
IP: "0.0.0.1",
},
{
IP: "0.0.0.2",
},
{
IP: "0.0.0.3",
},
h.OnAdd(&discoveryv1.EndpointSlice{
Endpoints: []discoveryv1.Endpoint{
{
Addresses: []string{"0.0.0.1"},
},
{
Addresses: []string{"0.0.0.2"},
},
{
Addresses: []string{"0.0.0.3"},
},
},
}}, false)
}, false)
assert.ElementsMatch(t, []string{"0.0.0.1", "0.0.0.2", "0.0.0.3"}, endpoints)
}
@@ -37,34 +35,30 @@ func TestDelete(t *testing.T) {
h := NewEventHandler(func(change []string) {
endpoints = change
})
h.OnAdd(&v1.Endpoints{Subsets: []v1.EndpointSubset{
{
Addresses: []v1.EndpointAddress{
{
IP: "0.0.0.1",
},
{
IP: "0.0.0.2",
},
{
IP: "0.0.0.3",
},
h.OnAdd(&discoveryv1.EndpointSlice{
Endpoints: []discoveryv1.Endpoint{
{
Addresses: []string{"0.0.0.1"},
},
{
Addresses: []string{"0.0.0.2"},
},
{
Addresses: []string{"0.0.0.3"},
},
},
}}, false)
}, false)
h.OnDelete("bad")
h.OnDelete(&v1.Endpoints{Subsets: []v1.EndpointSubset{
{
Addresses: []v1.EndpointAddress{
{
IP: "0.0.0.1",
},
{
IP: "0.0.0.2",
},
h.OnDelete(&discoveryv1.EndpointSlice{
Endpoints: []discoveryv1.Endpoint{
{
Addresses: []string{"0.0.0.1"},
},
{
Addresses: []string{"0.0.0.2"},
},
},
}})
})
assert.ElementsMatch(t, []string{"0.0.0.3"}, endpoints)
}
@@ -73,36 +67,28 @@ func TestUpdate(t *testing.T) {
h := NewEventHandler(func(change []string) {
endpoints = change
})
h.OnUpdate(&v1.Endpoints{
Subsets: []v1.EndpointSubset{
h.OnUpdate(&discoveryv1.EndpointSlice{
Endpoints: []discoveryv1.Endpoint{
{
Addresses: []v1.EndpointAddress{
{
IP: "0.0.0.1",
},
{
IP: "0.0.0.2",
},
},
Addresses: []string{"0.0.0.1"},
},
{
Addresses: []string{"0.0.0.2"},
},
},
ObjectMeta: metav1.ObjectMeta{
ResourceVersion: "1",
},
}, &v1.Endpoints{
Subsets: []v1.EndpointSubset{
}, &discoveryv1.EndpointSlice{
Endpoints: []discoveryv1.Endpoint{
{
Addresses: []v1.EndpointAddress{
{
IP: "0.0.0.1",
},
{
IP: "0.0.0.2",
},
{
IP: "0.0.0.3",
},
},
Addresses: []string{"0.0.0.1"},
},
{
Addresses: []string{"0.0.0.2"},
},
{
Addresses: []string{"0.0.0.3"},
},
},
ObjectMeta: metav1.ObjectMeta{
@@ -116,33 +102,25 @@ func TestUpdateNoChange(t *testing.T) {
h := NewEventHandler(func(change []string) {
assert.Fail(t, "should not called")
})
h.OnUpdate(&v1.Endpoints{
Subsets: []v1.EndpointSubset{
h.OnUpdate(&discoveryv1.EndpointSlice{
Endpoints: []discoveryv1.Endpoint{
{
Addresses: []v1.EndpointAddress{
{
IP: "0.0.0.1",
},
{
IP: "0.0.0.2",
},
},
Addresses: []string{"0.0.0.1"},
},
{
Addresses: []string{"0.0.0.2"},
},
},
ObjectMeta: metav1.ObjectMeta{
ResourceVersion: "1",
},
}, &v1.Endpoints{
Subsets: []v1.EndpointSubset{
}, &discoveryv1.EndpointSlice{
Endpoints: []discoveryv1.Endpoint{
{
Addresses: []v1.EndpointAddress{
{
IP: "0.0.0.1",
},
{
IP: "0.0.0.2",
},
},
Addresses: []string{"0.0.0.1"},
},
{
Addresses: []string{"0.0.0.2"},
},
},
ObjectMeta: metav1.ObjectMeta{
@@ -156,45 +134,35 @@ func TestUpdateChangeWithDifferentVersion(t *testing.T) {
h := NewEventHandler(func(change []string) {
endpoints = change
})
h.OnAdd(&v1.Endpoints{Subsets: []v1.EndpointSubset{
{
Addresses: []v1.EndpointAddress{
{
IP: "0.0.0.1",
},
{
IP: "0.0.0.3",
},
h.OnAdd(&discoveryv1.EndpointSlice{
Endpoints: []discoveryv1.Endpoint{
{
Addresses: []string{"0.0.0.1"},
},
{
Addresses: []string{"0.0.0.3"},
},
},
}}, false)
h.OnUpdate(&v1.Endpoints{
Subsets: []v1.EndpointSubset{
}, false)
h.OnUpdate(&discoveryv1.EndpointSlice{
Endpoints: []discoveryv1.Endpoint{
{
Addresses: []v1.EndpointAddress{
{
IP: "0.0.0.1",
},
{
IP: "0.0.0.3",
},
},
Addresses: []string{"0.0.0.1"},
},
{
Addresses: []string{"0.0.0.3"},
},
},
ObjectMeta: metav1.ObjectMeta{
ResourceVersion: "1",
},
}, &v1.Endpoints{
Subsets: []v1.EndpointSubset{
}, &discoveryv1.EndpointSlice{
Endpoints: []discoveryv1.Endpoint{
{
Addresses: []v1.EndpointAddress{
{
IP: "0.0.0.1",
},
{
IP: "0.0.0.2",
},
},
Addresses: []string{"0.0.0.1"},
},
{
Addresses: []string{"0.0.0.2"},
},
},
ObjectMeta: metav1.ObjectMeta{
@@ -209,63 +177,49 @@ func TestUpdateNoChangeWithDifferentVersion(t *testing.T) {
h := NewEventHandler(func(change []string) {
endpoints = change
})
h.OnAdd(&v1.Endpoints{Subsets: []v1.EndpointSubset{
{
Addresses: []v1.EndpointAddress{
{
IP: "0.0.0.1",
},
{
IP: "0.0.0.2",
},
},
},
}}, false)
h.OnUpdate("bad", &v1.Endpoints{Subsets: []v1.EndpointSubset{
{
Addresses: []v1.EndpointAddress{
{
IP: "0.0.0.1",
},
},
},
}})
h.OnUpdate(&v1.Endpoints{Subsets: []v1.EndpointSubset{
{
Addresses: []v1.EndpointAddress{
{
IP: "0.0.0.1",
},
},
},
}}, "bad")
h.OnUpdate(&v1.Endpoints{
Subsets: []v1.EndpointSubset{
h.OnAdd(&discoveryv1.EndpointSlice{
Endpoints: []discoveryv1.Endpoint{
{
Addresses: []v1.EndpointAddress{
{
IP: "0.0.0.1",
},
{
IP: "0.0.0.2",
},
},
Addresses: []string{"0.0.0.1"},
},
{
Addresses: []string{"0.0.0.2"},
},
},
}, false)
h.OnUpdate("bad", &discoveryv1.EndpointSlice{
Endpoints: []discoveryv1.Endpoint{
{
Addresses: []string{"0.0.0.1"},
},
},
})
h.OnUpdate(&discoveryv1.EndpointSlice{
Endpoints: []discoveryv1.Endpoint{
{
Addresses: []string{"0.0.0.1"},
},
},
}, "bad")
h.OnUpdate(&discoveryv1.EndpointSlice{
Endpoints: []discoveryv1.Endpoint{
{
Addresses: []string{"0.0.0.1"},
},
{
Addresses: []string{"0.0.0.2"},
},
},
ObjectMeta: metav1.ObjectMeta{
ResourceVersion: "1",
},
}, &v1.Endpoints{
Subsets: []v1.EndpointSubset{
}, &discoveryv1.EndpointSlice{
Endpoints: []discoveryv1.Endpoint{
{
Addresses: []v1.EndpointAddress{
{
IP: "0.0.0.1",
},
{
IP: "0.0.0.2",
},
},
Addresses: []string{"0.0.0.1"},
},
{
Addresses: []string{"0.0.0.2"},
},
},
ObjectMeta: metav1.ObjectMeta{

51
zrpc/resolver/internal/kubebuilder.go
View File

@@ -18,8 +18,8 @@ import (
)
const (
resyncInterval = 5 * time.Minute
nameSelector = "metadata.name="
resyncInterval = 5 * time.Minute
serviceSelector = "kubernetes.io/service-name="
)
type kubeResolver struct {
@@ -60,14 +60,33 @@ func (b *kubeBuilder) Build(target resolver.Target, cc resolver.ClientConn,
}
if svc.Port == 0 {
// getting endpoints is only to get the port
endpoints, err := cs.CoreV1().Endpoints(svc.Namespace).Get(
context.Background(), svc.Name, v1.GetOptions{})
endpointSlices, err := cs.DiscoveryV1().EndpointSlices(svc.Namespace).List(context.Background(),
v1.ListOptions{
LabelSelector: serviceSelector + svc.Name,
})
if err != nil {
return nil, err
}
if len(endpointSlices.Items) == 0 {
return nil, fmt.Errorf("no endpoint slices found for service %s in namespace %s",
svc.Name, svc.Namespace)
}
svc.Port = int(endpoints.Subsets[0].Ports[0].Port)
// Find the first slice with a valid port.
// Since this resolver is used for in-cluster service discovery,
// we expect at least one port to be available.
var foundPort bool
for _, slice := range endpointSlices.Items {
if len(slice.Ports) > 0 && slice.Ports[0].Port != nil {
svc.Port = int(*slice.Ports[0].Port)
foundPort = true
break
}
}
if !foundPort {
return nil, fmt.Errorf("no valid port found in endpoint slices for service %s in namespace %s",
svc.Name, svc.Namespace)
}
}
handler := kube.NewEventHandler(func(endpoints []string) {
@@ -88,23 +107,29 @@ func (b *kubeBuilder) Build(target resolver.Target, cc resolver.ClientConn,
inf := informers.NewSharedInformerFactoryWithOptions(cs, resyncInterval,
informers.WithNamespace(svc.Namespace),
informers.WithTweakListOptions(func(options *v1.ListOptions) {
options.FieldSelector = nameSelector + svc.Name
options.LabelSelector = serviceSelector + svc.Name
}))
in := inf.Core().V1().Endpoints()
in := inf.Discovery().V1().EndpointSlices()
_, err = in.Informer().AddEventHandler(handler)
if err != nil {
return nil, err
}
// get the initial endpoints, cannot use the previous endpoints,
// because the endpoints may be updated before/after the informer is started.
endpoints, err := cs.CoreV1().Endpoints(svc.Namespace).Get(
context.Background(), svc.Name, v1.GetOptions{})
// get the initial endpoint slices, cannot use the previous endpoint slices,
// because the endpoint slices may be updated before/after the informer is started.
endpointSlices, err := cs.DiscoveryV1().EndpointSlices(svc.Namespace).List(
context.Background(), v1.ListOptions{
LabelSelector: serviceSelector + svc.Name,
})
if err != nil {
return nil, err
}
handler.Update(endpoints)
// Aggregate endpoints from all EndpointSlices.
// Use OnAdd (not Update) to accumulate addresses across multiple slices.
for _, endpointSlice := range endpointSlices.Items {
handler.OnAdd(&endpointSlice, false)
}
r := &kubeResolver{
cc: cc,